"""DNS self-service request lifecycle. This service backs ``/api/v1/cmdb/dns-record-requests`` (employee submits) and the ``approval_type == "dns_record_request"`` branch of the Feishu callback dispatcher in ``app/approval/service.py`` (Feishu calls back when an admin approves / rejects). State machine (mirrors :mod:`app.domains.cmdb.models` constants): pending ──┬─▶ approved ─▶ executing ──┬─▶ executed │ └─▶ failed ├─▶ rejected └─▶ cancelled (requester withdrew before approval) The service intentionally reuses :class:`CmdbService` and the cloud provider classes for the actual DNS write — the only "new code on the hot path" is the validation + Feishu submission + state transitions. This means once-an-employee-submits-a-DNS-request, the bytes that reach Tencent DNSPod / Route53 / Aliyun / etc. are exactly the same bytes that already reach those APIs from the admin /admin/cloud/dns page that has been in production since PR #37. """ from __future__ import annotations import json import re import uuid as _uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.cancel_desync import notify_feishu_cancel_desync from app.approval.feishu_client import FeishuApprovalClient from app.approval.models import ApprovalRequest, ApprovalSubmissionIntent from app.approval.submission_intents import ( compact_terminal_intent_payload, defer_submission_intent_retry, reconciliation_due_predicate, reconciliation_order, submission_fingerprint, submit_with_durable_intent, terminal_intent_resource_id, ) from app.audit.service import AuditService from app.auth.models import AuthUser from app.core.exceptions import ( AppException, NotFoundError, ValidationError, ) from app.core.locale import is_en as locale_is_en from app.core.safe_logging import get_safe_logger from app.core.txn import best_effort from app.domains.cmdb.cert_service import fqdn_for from app.domains.cmdb.credential_crypto import decrypt_fields from app.domains.cmdb.models import ( DNS_REQ_ACTION_CREATE, DNS_REQ_ACTION_DELETE, DNS_REQ_ACTION_DISABLE, DNS_REQ_ACTION_ENABLE, DNS_REQ_ACTION_UPDATE, DNS_REQ_ACTIONS, DNS_REQ_APPROVED, DNS_REQ_CANCELLED, DNS_REQ_EXECUTED, DNS_REQ_EXECUTING, DNS_REQ_FAILED, DNS_REQ_PENDING, DNS_REQ_REJECTED, DNS_REQ_SELF_SERVICE_TYPES, DNS_REQ_TERMINAL_STATES, DNS_REQ_VALUE_ACTIONS, CmdbDnsRecord, CmdbDnsRecordRequest, CmdbDomain, ) from app.domains.cmdb.providers.base import ( DnsRecordWrite, ProviderCredentials, get_provider_class, min_dns_ttl, ) from app.domains.cmdb.schemas import ( DnsRecordRequestCreate, DnsRecordRequestResponse, ) from app.domains.cmdb.service import CmdbService from app.identity.person_models import IdentityPerson from app.integrations.compat import get_feishu_config from app.integrations.models import Integration from app.integrations.service import get_approval_flow logger = get_safe_logger(__name__) def _display_fqdn(name: str, zone_name: str) -> str: """Human-facing FQDN for approval forms / Feishu cards / error text. Apex records use ``name == "@"`` (see schemas); naive ``f"{name}.{zone}"`` renders that as ``@.example.com`` — wrong, and an approver decides on what they see. Reuse the canonical :func:`app.domains.cmdb.cert_service.fqdn_for` (``@``/empty/==zone → bare zone; already-FQDN names pass through). ``fqdn_for`` returns ``None`` for wildcard (``*``) because it's probe-oriented and ``*`` isn't probeable — for *display* a wildcard is meaningful, so fall back to the literal ``name.zone`` (e.g. ``*.example.com``). """ return fqdn_for(name, zone_name) or f"{name}.{zone_name}" # ─── Validation helpers ─────────────────────────────────────────── # IPv4 dotted-quad. Slightly permissive (we don't reject 999.x.x.x at # the regex level) — Tencent / Route 53 will reject the bad ones at # write time, and we'd rather surface that error than maintain a full # IP validator here. The job is "is this even shaped like an IP", not # "is this routable". _IPV4_RE = re.compile(r"^(\d{1,3}\.){3}\d{1,3}$") # IPv6: just check for hex segments + colons. Same trust-the-cloud # approach as IPv4. _IPV6_RE = re.compile(r"^[0-9a-fA-F:]+$") # Hostname / FQDN — letters, digits, underscores, hyphens, dots, max 253 # chars per spec. We allow trailing dot for canonical CNAME format. # Underscore-led labels are accepted here because cloud-issued validation # targets routinely use them — e.g. AWS ACM DNS validation CNAMEs point at # ``_b69bdcc...jkddzztszm.acm-validations.aws.`` (both the name AND the # target start with ``_``). The TLD label stays letters-only. Hyphen still # can't lead/trail a label. This is a *target* value (where the record # points), not the name being created, so the reserved-underscore denylist # (which guards the ``name`` column) doesn't apply. _HOSTNAME_RE = re.compile( r"^([a-zA-Z0-9_]([a-zA-Z0-9_\-]{0,61}[a-zA-Z0-9_])?\.)+[a-zA-Z]{2,}\.?$", ) # Sub-domain part (the "name" column) — RFC 1035 strict is letter-led # alphanumeric+hyphen, but RFC 8552 + universal cloud-DNS practice # (Aliyun DNS, AWS Route 53, Cloudflare, Google Cloud DNS) accept # underscore-led labels too — that's the cross-vendor convention for # service-discovery records (``_dmarc``, ``_acme-challenge``, # ``_dnsauth`` for domain-ownership verification, SRV bindings like # ``_sip._tcp``). Rejecting them at the regex layer would diverge # Helix from every provider's actual policy. # # Allowed chars per label: ``[A-Za-z0-9_-]``. Hyphen STILL not allowed # as first/last char (RFC 1035 — Aliyun rejects this too). Underscore # at first/last is fine. ``@`` (apex) and ``*`` (wildcard) are # whole-name shorthands handled outside the regex. # # Wildcard (``*``) is INTENTIONALLY NOT allowed for self-service: a # wildcard create hands the attacker every previously-unbound # subdomain of the zone in one approval. Admins who legitimately need # a wildcard record must use the admin direct-write path # (``POST /dns-records`` with ``can_manage_cloud``). High-risk reserved # underscore labels (``_acme-challenge`` / ``_dmarc`` / SPF / DKIM # etc — see ``_RESERVED_UNDERSCORE_LABELS`` below) are still # admin-only via the per-label denylist, which now actually gets # exercised (the prior regex rejected every underscore-led label # before the denylist could run, making the denylist dead code). _SUBDOMAIN_RE = re.compile( r"^(@|([A-Za-z0-9_]([A-Za-z0-9_\-]{0,61}[A-Za-z0-9_])?)" r"(\.[A-Za-z0-9_]([A-Za-z0-9_\-]{0,61}[A-Za-z0-9_])?)*)$", ) # Underscore-prefixed labels reserved for service records — SPF / DKIM / # DMARC / ACME-DNS-01 cert issuance / SMTPS TLSA. Self-service requests # for these would let an employee hijack email-auth or issue an # attacker-controlled cert; require admin direct-write instead. _RESERVED_UNDERSCORE_LABELS = { "_acme-challenge", "_dmarc", "_spf", "_dkim", "_domainkey", "_smtp", "_smtps", "_imap", "_imaps", "_pop3", "_pop3s", "_mta-sts", "_tls", "_tlsa", "_smimecert", } # TXT values that resemble email-auth or cert-issuance policy records. # Reject any TXT containing these prefixes from the self-service path — # an employee should never be able to put a v=spf1 / v=DMARC1 / v=DKIM1 # on apex or any subdomain via approval flow. Admin direct-write keeps # the option for legit operations. _TXT_RESERVED_PREFIXES = ( "v=spf1", "v=dmarc1", "v=dkim1", "google-site-verification=", "ms=", # Microsoft domain ownership "apple-domain-verification=", "atlassian-domain-verification=", ) def _validate_value_for_type(record_type: str, value: str) -> None: """Reject obvious garbage at submit time so the form catches it before a Feishu approval gets created. The cloud provider does the final word — this is just so a typo doesn't reach an admin's Feishu inbox.""" value = value.strip() if not value: raise ValidationError("Resolution target value is required.") if record_type == "A": if not _IPV4_RE.match(value): raise ValidationError( "A record value must be an IPv4 address (e.g. 1.2.3.4).", ) elif record_type == "AAAA": if not _IPV6_RE.match(value): raise ValidationError( "AAAA record value must be an IPv6 address.", ) elif record_type == "CNAME": if not _HOSTNAME_RE.match(value): raise ValidationError( "CNAME value must be a hostname (e.g. target.example.com).", ) elif record_type == "NS": # NS delegates a (sub)zone to an authoritative nameserver — the # value is that nameserver's FQDN, same shape as a CNAME target. if not _HOSTNAME_RE.match(value): raise ValidationError( "NS record value must be a hostname (e.g. ns1.example.com).", ) elif record_type == "TXT": if len(value) > 4096: raise ValidationError("TXT value exceeds 4096 chars.") # Round-28 (2026-05-18): reject email-auth / domain-ownership # TXT values from self-service. Pure case-insensitive prefix # match — the admin direct-write path is unaffected because it # never calls this function. Iterative leading-junk strip: # mixed quote orderings (e.g. ``\'"v=spf1..."``) would bypass # a single-pass lstrip — keep peeling until the next character # is the actual policy keyword. lowered = value.lower() _JUNK = ' "\'\t\n\r' while lowered and lowered[0] in _JUNK: lowered = lowered[1:] for reserved in _TXT_RESERVED_PREFIXES: if lowered.startswith(reserved): raise ValidationError( "TXT records carrying SPF / DMARC / DKIM / " "domain-ownership tokens must be created by an admin via " "the direct-write path. Self-service is intentionally " "not allowed for these — they grant email-auth or " "external-service authority over the zone.", ) else: # Defense in depth — schema literal should catch this earlier. raise ValidationError( f"Record type {record_type!r} not allowed for self-service.", ) def _public_error_message(exc: BaseException) -> str: """Build a generic, non-leaky failure message safe to expose to the DNS request requester via ``DnsRecordRequestResponse.error_message``. The raw provider exception typically carries AccessKeyId fragments, ARNs, account IDs, region names, and signed URL pieces — exposing those through the employee-facing GET would leak provider topology. Whitelist a small set of provider classifications by exception-class name; fall back to a generic message. """ name = type(exc).__name__ # Common provider-SDK classifications. Names are stable enough # across boto/aliyun/qcloud SDK versions to key on. if "AccessDenied" in name or "Unauthorized" in name or "Forbidden" in name: return ( "Cloud provider rejected the DNS write: access denied. " "Admin: check integration credentials + IAM scope." ) if "Throttling" in name or "RateLimit" in name or "TooManyRequests" in name: return "Cloud provider throttled the DNS write — retry shortly." if "Timeout" in name or "ConnectionError" in name or "ServiceUnavailable" in name: return "Cloud provider DNS API is unreachable — retry shortly." if "NotFound" in name or "NoSuchHostedZone" in name: return ( "Cloud provider could not find the target record / zone. " "It may have been deleted or moved between submit and approval." ) if "Conflict" in name or "AlreadyExists" in name or "Duplicate" in name: return ( "Cloud provider reports a conflicting record already exists. " "Refresh and try an update / delete instead." ) if "Validation" in name or "InvalidParameter" in name or "BadRequest" in name: return ( "Cloud provider rejected the DNS payload as malformed. " "Admin: check the audit log for the exact provider error." ) return ( "Cloud DNS write failed. The admin team has been notified — " "the full error is recorded in the audit log." ) def _validate_subdomain(name: str) -> None: # Wildcard rejected upfront — same wording the user sees on the # Feishu approval card. ``*.zone`` creates would let the requester # silently bind every unbound subdomain to a value of their choosing. if "*" in name: raise ValidationError( "Wildcard hostnames (``*``) are not allowed via self-service. " "Submit one record per concrete subdomain, or ask an admin " "to write the wildcard directly.", ) if not _SUBDOMAIN_RE.match(name): raise ValidationError( "Hostname contains illegal characters. Use letters, digits, " "underscores, hyphens, dots, '@', or '*'. Hyphens cannot " "lead or trail a label.", ) # Reject reserved underscore labels in ANY position — SPF / DKIM / # DMARC / ACME-DNS-01 / MTA-STS. Bare label (e.g. ``_dmarc``) and # subdomain forms (``_acme-challenge.www`` OR ``www._acme-challenge``) # are both blocked. Real-world phishing-style typo-squat plus # operator confusion both motivate the broader check. labels = {part.lower() for part in name.split(".")} blocked = labels & _RESERVED_UNDERSCORE_LABELS if blocked: bad = sorted(blocked)[0] raise ValidationError( f"Hostname label {bad!r} is reserved for email-authentication " "/ cert-issuance service records. Ask an admin to write this " "record directly — self-service is intentionally not allowed " "for these labels.", ) # ─── Provider factory ───────────────────────────────────────────── async def _load_dns_provider( db: AsyncSession, integration_id: str, ): """Build a live cloud provider client for the given integration. Mirrors ``_load_provider_for_dns`` in ``cmdb/router.py`` (same field names, same archive-tenant tolerance) — duplicated here so the callback handler doesn't import the router (which would create a circular dependency through approval/service.py). """ integration = await db.get(Integration, _uuid.UUID(str(integration_id))) if integration is None or integration.kind != "cloud": raise NotFoundError("CloudIntegration", str(integration_id)) archival_providers = {"ksyun", "ucloud"} if ( integration.provider in archival_providers and not integration.secrets_encrypted ): # Archive-only tenant — instantiate a provider stub so any DNS # write surfaces a CapabilityNotSupported, not a generic 404. creds = ProviderCredentials( provider_id=str(integration.id), provider_type=integration.provider, account_alias=integration.name, fields={}, ) return get_provider_class(integration.provider)(creds) if not integration.secrets_encrypted: raise NotFoundError("CloudIntegration", str(integration_id)) cred_fields = decrypt_fields( integration.secrets_encrypted, integration.secrets_nonce or "", integration_id=str(integration.id), ) merged: dict[str, str] = {} if integration.config_public: merged.update( {k: str(v) for k, v in integration.config_public.items() if v is not None}, ) merged.update(cred_fields) creds = ProviderCredentials( provider_id=str(integration.id), provider_type=integration.provider, account_alias=integration.name, fields=merged, ) return get_provider_class(integration.provider)(creds) # ─── Feishu form serialization helpers ──────────────────────────── # Widget-name keyword fragments → field-value key. Substring match, # case-insensitive, against the Feishu template's widget name. The # admin's approval template can use whatever IDs / names they like — # we only need ANY of the listed keywords to appear inside the name. # # Iteration order matters — Python preserves insertion order for dicts, # and ``_build_dns_form_payload`` matches against the FIRST field whose # keyword set hits. ``current_value`` is intentionally listed BEFORE # ``value`` so a widget named "Current (Old) parsing value" matches the # old-value field on its "old" / "原值" cue rather than getting grabbed # by ``value``'s broader "value" / "记录值" keywords. _DNS_FIELD_KEYWORDS: dict[str, tuple[str, ...]] = { "action": ("操作", "action", "类型", "kind", "operation"), "fqdn": ("完整", "fqdn", "完整主机", "全名"), "zone": ("zone", "域名", "domain"), "name": ("主机", "host", "subdomain", "记录名"), "current_value": ("当前值", "current value", "原值", "原解析", "old"), "record_type": ("记录类型", "record type", "类型 type", "type"), "value": ("解析值", "解析目标", "value", "target", "记录值"), "ttl": ("ttl", "生存"), "purpose": ("用途", "purpose", "原因", "reason", "说明"), } # Widget types that submit-side should skip outright (file uploads, # sectioning headers, etc.). Mirrors ApiKeyService._SKIP_WIDGET_TYPES. _SKIP_WIDGET_TYPES = frozenset({"text", "section", "image", "attachmentV2"}) def _parse_template_widgets(raw_form: Any) -> list[dict[str, Any]]: if isinstance(raw_form, str): return json.loads(raw_form or "[]") if isinstance(raw_form, list): return raw_form return [] def _match_widget(widget_name: str, field_key: str) -> bool: """Return True if widget name contains any keyword for the field.""" name_lower = (widget_name or "").lower() keywords = _DNS_FIELD_KEYWORDS.get(field_key, ()) return any(kw.lower() in name_lower for kw in keywords) def _build_dns_form_payload( widgets: list[dict[str, Any]], values: dict[str, Any], ) -> str: """Walk the admin's Feishu template widgets, attach the matching DNS field value to each one, drop unmatched / required-but-empty widgets cleanly, and return the JSON string Feishu wants. Honours Feishu's ``display_condition`` via the shared evaluator in ``app.integrations.feishu_form_visibility`` — hidden widgets are skipped entirely so admins can mark a follow-up widget "required when type=CNAME" without breaking submits where the trigger isn't met. Fail-open on any unknown schema variant. """ from app.integrations.feishu_form_visibility import ( evaluate_display_conditions, ) visibility = evaluate_display_conditions(widgets, {}, assume_visible_when_answers_unsourced=True) serialized: list[dict[str, Any]] = [] for widget in widgets: wtype = widget.get("type", "input") if wtype in _SKIP_WIDGET_TYPES: continue wid = widget.get("id") if not wid: continue if not visibility.get(wid, True): continue wname = widget.get("name") or "" # Find the first field whose keyword set matches this widget's # name. Two widgets matching the same field is unusual but # both will get the same value if it happens. matched_value = None for field_key in _DNS_FIELD_KEYWORDS: if _match_widget(wname, field_key) and field_key in values: matched_value = values[field_key] break if matched_value is None: # Fall back to "purpose" for any leftover required textarea # so the admin reading the approval at least sees the # employee's stated reason rather than an empty form. if widget.get("required") and wtype in ("textarea", "input"): matched_value = values.get("purpose", "") else: continue if matched_value in (None, "", [], {}): continue text = ( str(matched_value) if not isinstance(matched_value, (dict, list)) else json.dumps(matched_value, ensure_ascii=False) ) serialized.append({"id": wid, "type": wtype, "value": text}) return json.dumps(serialized, ensure_ascii=False) # ─── Service ────────────────────────────────────────────────────── _FEISHU_FLOW_KEY = "dns_record_request" # Stable fragment the resume gate matches on. Kept separate from the full # sentence so the wording can be reworded / translated without silently # making already-failed rows un-resumable. _STRANDED_EXECUTION_MARKER = "execution was interrupted" STRANDED_EXECUTION_ERROR = ( "Execution was interrupted before Helix could confirm the result, and " "retrying kept getting interrupted. The DNS record was NOT changed — " "verified against the provider. The approval is still valid: an IT " "admin can resume this request." ) class DnsRequestService: """Submit + cancel + list + Feishu-callback-execute. One instance per request — it caches the Feishu client lazily so a list-only call doesn't hit the DB twice. """ # Outcome of the most recent ``cancel()`` call's best-effort Feishu # side-cancel. Routers / MCP tools read it after ``cancel()`` to # surface ``feishu_cancel_synced`` to the requester. last_feishu_cancel_synced: bool = True def __init__(self, db: AsyncSession): self.db = db self._feishu_client: FeishuApprovalClient | None = None async def _client(self) -> FeishuApprovalClient: if self._feishu_client is None: cfg = await get_feishu_config(session=self.db) self._feishu_client = FeishuApprovalClient( app_id=str(cfg.get("app_id") or ""), app_secret=str(cfg.get("app_secret") or ""), ) return self._feishu_client # ── owner-label helper (used by list + response builder) ─── async def _label_for_person(self, person_id: str | None) -> str | None: if not person_id: return None person = await self.db.get(IdentityPerson, person_id) if not person: return None return person.display_name or person.work_email or None async def _to_response( self, row: CmdbDnsRecordRequest, ) -> DnsRecordRequestResponse: return DnsRecordRequestResponse( id=row.id, requester_person_id=str(row.requester_person_id), requester_label=await self._label_for_person(row.requester_person_id), action=row.action, # type: ignore[arg-type] integration_id=row.integration_id, zone_name=row.zone_name, name=row.name, record_type=row.record_type, value=row.value, ttl=row.ttl, line=row.line, target_record_id=str(row.target_record_id) if row.target_record_id else None, purpose=row.purpose, attachment_url=row.attachment_url, status=row.status, # type: ignore[arg-type] feishu_instance_code=row.feishu_instance_code, error_message=row.error_message, submitted_at=row.submitted_at, decided_at=row.decided_at, executed_at=row.executed_at, executed_record_pk=str(row.executed_record_pk) if row.executed_record_pk else None, created_at=row.created_at, updated_at=row.updated_at, ) # ── Validation step (called from submit) ─────────────────── async def _resolve_target_record( self, body: DnsRecordRequestCreate, requester: AuthUser, ) -> CmdbDnsRecord | None: """For update/delete, fetch the target record + assert ownership. Returns None for ``create`` (no target). """ if body.action == DNS_REQ_ACTION_CREATE: if body.target_record_id is not None: raise ValidationError( "target_record_id must be null when action='create'.", ) return None if body.target_record_id is None: raise ValidationError( f"target_record_id is required when action='{body.action}'.", ) target = await self.db.get( CmdbDnsRecord, _uuid.UUID(str(body.target_record_id)), ) if target is None or target.deleted_at is not None: raise NotFoundError("DnsRecord", str(body.target_record_id)) if str(target.owner_person_id or "") != str(requester.id): raise AppException( code="DNS_REQ_NOT_OWNED", message=( "You can only modify or delete DNS records you own." " Contact IT to change ownership first." ), status_code=403, ) if target.record_type not in DNS_REQ_SELF_SERVICE_TYPES: raise AppException( code="DNS_REQ_TYPE_NOT_SELF_SERVICE", message=( f"Record type {target.record_type} is admin-only and" " cannot be modified through self-service." ), status_code=400, ) # Same-zone same-integration sanity check — UI should enforce # this but defense in depth in case someone calls the API # directly. if str(target.integration_id) != str(body.integration_id): raise ValidationError( "integration_id does not match the target record.", ) if target.zone_name != body.zone_name: raise ValidationError("zone_name does not match the target record.") return target async def _check_create_conflict( self, body: DnsRecordRequestCreate, ) -> None: """Reject a ``create`` that would clash with an existing record. Catching this at submit time avoids the worst UX failure mode: admin approves the Feishu, callback runs, Tencent rejects with ``RECORD_EXISTS`` and the request lands in ``failed``. Cheaper to fail fast. """ if body.action != DNS_REQ_ACTION_CREATE: return # (zone, name, type) is NOT unique in ``cmdb_dns_records`` — # multi-value rrsets (an NS delegation's 2-4 nameservers, MX # sets, multiple TXT verification tokens) sync as one row per # value. Any-row-exists is the conflict signal; asking for # exactly one (``scalar_one_or_none``) raised # ``MultipleResultsFound`` → 500 on every such name. result = await self.db.execute( select(CmdbDnsRecord.id).where( and_( CmdbDnsRecord.integration_id == str(body.integration_id), CmdbDnsRecord.zone_name == body.zone_name, CmdbDnsRecord.name == body.name, CmdbDnsRecord.record_type == body.record_type, CmdbDnsRecord.deleted_at.is_(None), ), ).limit(1), ) if result.scalar() is not None: raise AppException( code="DNS_REQ_RECORD_EXISTS", message=( f"A {body.record_type} record for " f"{_display_fqdn(body.name, body.zone_name)} already " "exists. Submit an 'update' request instead." ), status_code=409, ) async def _validate_ttl_floor( self, *, integration_id: str, provider: str | None, ttl: int, ) -> None: """Reject a TTL the DNS provider is guaranteed to refuse. Same "fail fast, not after approval" motive as :meth:`_check_create_conflict`, learned the expensive way: a pingcap.net CNAME submitted with TTL=300 cleared the schema's ``ge=60`` bound, cleared three levels of human approval, and only then hit DNSPod's ``LimitExceeded.RecordTtlLimit``. The requester waited a day to be told a number they could have fixed instantly. Deliberately a floor check only — no clamping. Silently rewriting the TTL would make the executed record differ from the value the approvers signed off on. """ integration = await self.db.get(Integration, str(integration_id)) floor = min_dns_ttl( provider, config_public=(integration.config_public if integration else None), ) if ttl >= floor: return raise ValidationError( f"TTL {ttl} is below the minimum this DNS provider accepts " f"({floor}). Set TTL to {floor} or higher and re-submit — the " f"provider rejects anything lower at write time, so an approved " f"request with TTL {ttl} could never be applied. " f"TTL {ttl} 低于该 DNS 服务商允许的最小值 {floor}，请改为 {floor} " f"或更大后重新提交。" ) # ── Submit ───────────────────────────────────────────────── async def submit( self, *, body: DnsRecordRequestCreate, requester: AuthUser, trace_id: str, ip_address: str | None, user_agent: str | None, ) -> DnsRecordRequestResponse: if body.action not in DNS_REQ_ACTIONS: raise ValidationError(f"Unsupported action {body.action!r}.") _validate_subdomain(body.name) # Value-format validation only applies to actions that actually # write a value (create / update). For enable / disable / delete # the ``value`` field is carried for display only — a stale or # oddly-formatted value (e.g. a previous CNAME target now # expired) shouldn't block re-enabling a record. if body.action in DNS_REQ_VALUE_ACTIONS: _validate_value_for_type(body.record_type, body.value) # Zone authority gate: a self-service DNS request MUST target a # zone the admin has explicitly left open via # ``cmdb_domains.allow_dns_request=True``. ``cmdb_domains.integration_id`` # is the **registrar** (not the DNS provider) — refer to the # 2026-06-01 PR diagnosis. Match on ``domain_name`` only; the # DNS provider identity is validated against # ``cmdb_dns_records`` further down. zone_row = await self.db.scalar( select(CmdbDomain).where( CmdbDomain.domain_name == body.zone_name, CmdbDomain.deleted_at.is_(None), ), ) if zone_row is None: raise ValidationError( f"No company-registered zone '{body.zone_name}' — open " "the 域名管理 list, verify the zone is synced, and " "re-submit." ) if not zone_row.allow_dns_request: raise ValidationError( f"Zone '{body.zone_name}' is not open for self-service " "DNS requests. Ask an admin to enable 「允许申请 DNS 解析」 " "on the cloud domain edit drawer." ) # DNS provider authority gate: the integration the user picked # MUST be the resolved authoritative-DNS provider for this zone # (i.e. it has at least one synced record in ``cmdb_dns_records``, # or it's the winner of the tie-break in # ``resolve_dns_integration_for_zone``). Without this gate, a # caller could POST ``integration_id=`` for a # zone that integration doesn't host, and the Feishu approval # would drive a write to the wrong cloud — silently failing # late or, worse, succeeding on the wrong tenant if the # wrong tenant happens to also host that zone in shadow. from app.domains.cmdb.service import CmdbService # local: avoid cycle resolved = await CmdbService.resolve_dns_integration_for_zone( self.db, body.zone_name, ) if resolved is None: # No synced record exists for this zone on any enabled # provider — we can't safely route the write. Admin must # seed the first record from the cloud console (so sync # discovers the zone-provider mapping), then self-service # opens up. raise ValidationError( f"Zone '{body.zone_name}' has no synced DNS provider " "yet — IT Admin must seed the first record from the " "cloud console (sync will then discover the zone), " "after which self-service requests can route correctly." ) resolved_integ_id, resolved_integ_name, _resolved_provider = resolved if str(body.integration_id) != resolved_integ_id: raise ValidationError( f"Zone '{body.zone_name}' is authoritatively hosted by " f"'{resolved_integ_name}' (DNS provider id " f"{resolved_integ_id}); the request targeted " f"'{body.integration_id}'. The DNS picker should have " "pre-filled the right tenant — refresh the page and " "re-submit. If you're certain the picker is wrong, " "raise an IT Admin ticket: this usually means the " "DNS sync for that tenant hasn't run recently." ) # TTL floor gate. The provider — not Helix — owns the real lower # bound, and it enforces it at WRITE time, i.e. after the whole # approval chain has already run. Checking it here converts a # post-approval dead end into an instant form error the requester # can fix in one edit. Only value-carrying actions are gated: # delete / enable / disable never send a TTL. if body.action in DNS_REQ_VALUE_ACTIONS: await self._validate_ttl_floor( integration_id=resolved_integ_id, provider=_resolved_provider, ttl=body.ttl, ) target = await self._resolve_target_record(body, requester) await self._check_create_conflict(body) # Resolve approval flow (admin must have wired the Feishu code # under /admin/integrations//flows). We resolve before # creating the row so we don't write a half-baked request that # will never reach Feishu. flow = await get_approval_flow(_FEISHU_FLOW_KEY, session=self.db) approval_code = flow.external_code # Fetch the live Feishu template so we can match widget IDs by # widget name (admins design the template; we only know the # field meanings, not the random Feishu widget IDs). Same # pattern as ``ApiKeyService.create_api_key_request``. client = await self._client() definition = await client.get_approval_definition(approval_code) widgets = _parse_template_widgets(definition.get("form", "[]")) action_label = { DNS_REQ_ACTION_CREATE: "新增 / Create", DNS_REQ_ACTION_UPDATE: "修改 / Update", DNS_REQ_ACTION_DELETE: "删除 / Delete", DNS_REQ_ACTION_ENABLE: "启用 / Enable", DNS_REQ_ACTION_DISABLE: "停用 / Disable", }[body.action] # Field meanings the admin's template should expose. Matching # is by Chinese-or-English keyword fragments inside the widget # ``name`` — case-insensitive, substring. Templates that don't # carry one of these widgets just leave it blank (Feishu # accepts blank-but-not-required). field_values: dict[str, Any] = { "action": action_label, "fqdn": _display_fqdn(body.name, body.zone_name), "zone": body.zone_name, "name": body.name, "record_type": body.record_type, "value": body.value, "ttl": str(body.ttl), "purpose": body.purpose, } if target is not None: field_values["current_value"] = ( f"{target.value} (TTL {target.ttl})" ) form_payload = _build_dns_form_payload(widgets, field_values) # Map the requester's email to a Feishu user_id. ``work_email`` # is the canonical identifier between Helix and Feishu — every # other approval flow uses it the same way. feishu_user_id = await client.resolve_user_id_by_email(requester.work_email) semantic_payload = { "action": body.action, "integration_id": str(body.integration_id), "zone_name": body.zone_name, "name": body.name, "record_type": body.record_type, "value": body.value, "ttl": body.ttl, "target_record_id": str(target.id) if target is not None else None, "purpose": body.purpose, "attachment_url": body.attachment_url, } fingerprint = submission_fingerprint(semantic_payload) intent_scope = f"cmdb_dns:{fingerprint[:32]}" request_id = str(_uuid.uuid4()) approval_request_id = str(_uuid.uuid4()) from app.core.distributed_lock import LockBusy, distributed_lock try: async with distributed_lock( f"cmdb_dns_submit:{requester.id}:{fingerprint}", ttl=120, ): # A waiter may have opened a MySQL/TiDB RR snapshot before # acquiring Redis. FOR UPDATE is a current read, so it sees # the winner's committed row and turns a double-click into an # idempotent response instead of a second Feishu instance. existing = await self.db.scalar( select(CmdbDnsRecordRequest) .where( CmdbDnsRecordRequest.requester_person_id == str(requester.id), CmdbDnsRecordRequest.action == body.action, CmdbDnsRecordRequest.integration_id == str(body.integration_id), CmdbDnsRecordRequest.zone_name == body.zone_name, CmdbDnsRecordRequest.name == body.name, CmdbDnsRecordRequest.record_type == body.record_type, CmdbDnsRecordRequest.value == body.value, CmdbDnsRecordRequest.ttl == body.ttl, CmdbDnsRecordRequest.target_record_id == ( str(target.id) if target is not None else None ), CmdbDnsRecordRequest.purpose == body.purpose, CmdbDnsRecordRequest.attachment_url == body.attachment_url, CmdbDnsRecordRequest.status == DNS_REQ_PENDING, ) .order_by(CmdbDnsRecordRequest.created_at.desc()) .limit(1) .with_for_update() .execution_options(populate_existing=True) ) if existing is not None: return await self._to_response(existing) intent = await submit_with_durable_intent( self.db, client=client, scope=intent_scope, applicant_id=requester.id, semantic_payload=semantic_payload, provider_payload={ "approval_code": approval_code, "identity_type": "user_id", "identity_value": feishu_user_id, "form": form_payload, }, finalization_payload={ **semantic_payload, "mode": "submit", "request_id": request_id, "approval_request_id": approval_request_id, "approval_code": approval_code, }, trace_id=trace_id, ) row = await self._finalize_submission_intent( intent.id, recovered=intent.attempt_count > 1, ip_address=ip_address, user_agent=user_agent, ) return await self._to_response(row) except LockBusy as exc: raise AppException( code="DNS_REQ_SUBMIT_IN_FLIGHT", message=( "The same DNS request is already being submitted. " "Please retry in a few seconds." ), status_code=409, ) from exc async def _finalize_submission_intent( self, intent_id: _uuid.UUID, *, recovered: bool, ip_address: str | None = None, user_agent: str | None = None, ) -> CmdbDnsRecordRequest: """Atomically materialise DNS submit/retry state and its mirror.""" intent = await self.db.scalar( select(ApprovalSubmissionIntent) .where(ApprovalSubmissionIntent.id == intent_id) .with_for_update() .execution_options(populate_existing=True) ) if intent is None or not intent.scope.startswith("cmdb_dns"): raise RuntimeError("DNS submission intent disappeared") if intent.state == "completed": request_id = terminal_intent_resource_id( intent, resource_type="cmdb_dns_record_request", ) else: payload = dict((intent.payload or {}).get("finalization") or {}) request_id = str(payload["request_id"]) row = await self.db.scalar( select(CmdbDnsRecordRequest) .where(CmdbDnsRecordRequest.id == _uuid.UUID(request_id)) .with_for_update() .execution_options(populate_existing=True) ) if intent.state == "completed": if row is None: raise RuntimeError("completed DNS intent has no row") return row payload = dict((intent.payload or {}).get("finalization") or {}) if intent.state != "provider_created" or not intent.external_id: raise RuntimeError(f"cannot finalize DNS intent in {intent.state!r}") mode = str(payload.get("mode") or "") now = datetime.now(UTC) original_instance_code = "" original_error: str | None = None if mode == "submit": if row is None: row = CmdbDnsRecordRequest( id=_uuid.UUID(request_id), requester_person_id=str(intent.applicant_id), action=str(payload["action"]), integration_id=str(payload["integration_id"]), zone_name=str(payload["zone_name"]), name=str(payload["name"]), record_type=str(payload["record_type"]), value=str(payload["value"]), ttl=int(payload["ttl"]), line=None, target_record_id=payload.get("target_record_id"), purpose=str(payload["purpose"]), attachment_url=payload.get("attachment_url"), status=DNS_REQ_PENDING, feishu_instance_code=intent.external_id, submitted_at=now, ) self.db.add(row) await self.db.flush() elif mode == "retry": if row is None: raise RuntimeError("DNS retry intent lost its request row") if not self._is_four_eyes_failure(row): raise AppException( code="DNS_REQ_RETRY_STATE_DRIFT", message=( "The DNS request changed state while its retry was " "being reconciled; Helix will not overwrite it." ), status_code=409, ) original_instance_code = row.feishu_instance_code or "" original_error = row.error_message row.status = DNS_REQ_PENDING row.error_message = None row.decided_at = None row.feishu_instance_code = intent.external_id row.submitted_at = now else: raise RuntimeError(f"unknown DNS intent mode {mode!r}") approval_request = ApprovalRequest( id=_uuid.UUID(str(payload["approval_request_id"])), external_id=intent.external_id, approval_type=_FEISHU_FLOW_KEY, applicant_id=intent.applicant_id, status=DNS_REQ_PENDING, source_payload={ "dns_request_id": request_id, "approval_code": str(payload["approval_code"]), "action": str(payload["action"]), "zone_name": str(payload["zone_name"]), "name": str(payload["name"]), "record_type": str(payload["record_type"]), **( {"retry_of_instance_code": original_instance_code} if mode == "retry" else {} ), }, ) self.db.add(approval_request) await self.db.flush() row.approval_request_id = str(approval_request.id) intent.approval_request_id = approval_request.id intent.state = "completed" intent.completed_at = now intent.last_error = None await AuditService(self.db).log( trace_id=( f"approval-intent-reconcile:{intent.id}" if recovered else str((intent.payload or {}).get("trace_id")) ), actor_type="system" if recovered else "user", actor_user_id=( None if recovered else _uuid.UUID(str( payload.get("actor_user_id") or intent.applicant_id )) ), action=( "approval.submission_intent.reconciled" if recovered else ( "cmdb.dns_request.retried_after_block" if mode == "retry" else "cmdb.dns_request.submit" ) ), status="success", resource_type="cmdb_dns_record_request", resource_id=request_id, detail={ "intent_id": str(intent.id), "provider_request_uuid": intent.provider_request_uuid, "feishu_instance_code": intent.external_id, "recovered": recovered, "mode": mode, **( { "original_instance_code": original_instance_code, "original_error_message": original_error, } if mode == "retry" else {} ), }, ip_address=ip_address, user_agent=user_agent, ) compact_terminal_intent_payload( intent, resource_type="cmdb_dns_record_request", resource_id=request_id, ) await self.db.commit() await self.db.refresh(row) return row async def reconcile_submission_intents( self, *, min_age_minutes: int = 2, batch_limit: int = 100, ) -> dict[str, int]: """Recover DNS creates/retries without requiring an HTTP retry.""" cutoff = datetime.now(UTC) - timedelta(minutes=min_age_minutes) recoverable_states = ( "prepared", "submitting", "unknown", "provider_created", ) intent_ids = list((await self.db.scalars( select(ApprovalSubmissionIntent.id) .where( or_( ApprovalSubmissionIntent.scope.like( r"cmdb\_dns:%", escape="\\", ), ApprovalSubmissionIntent.scope.like( r"cmdb\_dns\_retry:%", escape="\\", ), ), ApprovalSubmissionIntent.state.in_(recoverable_states), ( (ApprovalSubmissionIntent.state == "provider_created") | (ApprovalSubmissionIntent.updated_at < cutoff) ), reconciliation_due_predicate(datetime.now(UTC)), ) .order_by(*reconciliation_order()) .limit(batch_limit) )).all()) summary = { "checked": len(intent_ids), "reconciled": 0, "still_unknown": 0, "failed": 0, } if not intent_ids: return summary client = await self._client() for intent_id in intent_ids: try: intent = await self.db.get(ApprovalSubmissionIntent, intent_id) if intent is None or intent.state == "completed": continue intent_payload = dict(intent.payload or {}) intent = await submit_with_durable_intent( self.db, client=client, scope=intent.scope, applicant_id=intent.applicant_id, semantic_payload=dict(intent_payload.get("semantic") or {}), provider_payload=dict(intent_payload.get("provider") or {}), finalization_payload=dict( intent_payload.get("finalization") or {} ), trace_id=str( intent_payload.get("trace_id") or f"approval-intent:{intent.id}" ), ) await self._finalize_submission_intent(intent.id, recovered=True) summary["reconciled"] += 1 except AppException as exc: if exc.code == "APPROVAL_SUBMISSION_RESULT_UNKNOWN": await self.db.rollback() summary["still_unknown"] += 1 else: await defer_submission_intent_retry(self.db, intent_id, exc) summary["failed"] += 1 except Exception as exc: # noqa: BLE001 await defer_submission_intent_retry(self.db, intent_id, exc) summary["failed"] += 1 return summary # ── List + Detail + Cancel ───────────────────────────────── async def list_for_user( self, *, requester: AuthUser, page: int = 1, page_size: int = 20, status: str | None = None, ) -> tuple[list[DnsRecordRequestResponse], int]: # Defensive bounds — the router-side ``Query(le=100)`` cap is # the primary defense; this clamp protects against MCP / # internal callers that bypass the FastAPI Query layer. if page_size > 100: page_size = 100 if page_size < 1: page_size = 1 if page < 1: page = 1 from sqlalchemy import func q = select(CmdbDnsRecordRequest).where( CmdbDnsRecordRequest.requester_person_id == str(requester.id), ) c = select(func.count()).select_from(CmdbDnsRecordRequest).where( CmdbDnsRecordRequest.requester_person_id == str(requester.id), ) if status: q = q.where(CmdbDnsRecordRequest.status == status) c = c.where(CmdbDnsRecordRequest.status == status) total = (await self.db.execute(c)).scalar() or 0 rows = (await self.db.execute( q.order_by(CmdbDnsRecordRequest.created_at.desc()) .offset((page - 1) * page_size).limit(page_size) )).scalars().all() return [await self._to_response(r) for r in rows], int(total) async def get( self, request_id: str, requester: AuthUser, ) -> DnsRecordRequestResponse: row = await self.db.get(CmdbDnsRecordRequest, _uuid.UUID(request_id)) if row is None: raise NotFoundError("DnsRecordRequest", request_id) # Self-fetch (the original requester) OR a cloud-admin / # global-admin acting on the broader admin centre. Without the # admin override, support staff who land on a deep-link from a # Feishu approval instance can't drill down to the DNS request # detail without going to the admin centre — usability gap. # ``can_manage_cloud`` covers it_admin and cloud_admin. from app.core.rbac import can_manage_cloud if str(row.requester_person_id) != str(requester.id) and \ not can_manage_cloud(requester.role_names): raise AppException( code="DNS_REQ_FORBIDDEN", message=( "You can only read your own DNS requests " "(global_admin / cloud_admin override applies)." ), status_code=403, ) return await self._to_response(row) async def cancel( self, *, request_id: str, requester: AuthUser, trace_id: str, ) -> DnsRecordRequestResponse: # Lock the row before the check-then-write so a concurrent # Feishu approval callback can't observe ``pending`` at the # same time and race us into a split-brain (user-cancel + # admin-approve both committing). Mirrors the pattern in the 5 # other generic-flow services (sso/domain/vpn/announcement/ # ai_tool) that all have an ``_lock_*`` helper. row = ( await self.db.execute( select(CmdbDnsRecordRequest) .where(CmdbDnsRecordRequest.id == _uuid.UUID(request_id)) .with_for_update() ) ).scalar_one_or_none() if row is None: raise NotFoundError("DnsRecordRequest", request_id) # Pre-launch P1 fix (Round 45 follow-up, CR-316): allow # cloud_admin / global_admin to cancel on behalf of the # requester. Mirrors the SSO grant flow's admin-on-behalf # cancel — without this, a leaver's pending DNS request # (the leaver-flip cascade now sweeps them; this also covers # the manual case where an admin spots the stale row first) # cannot be cleaned up without bypass-execute. from app.core.rbac import can_manage_cloud # noqa: PLC0415 is_owner = str(row.requester_person_id) == str(requester.id) is_admin = can_manage_cloud(requester.role_names) if not is_owner and not is_admin: raise AppException( code="DNS_REQ_FORBIDDEN", message="You can only cancel your own DNS requests.", status_code=403, ) if row.status != DNS_REQ_PENDING: raise AppException( code="DNS_REQ_NOT_CANCELLABLE", message=( f"Cannot cancel request in status {row.status!r};" " only pending requests can be cancelled." ), status_code=409, ) # Flip local state FIRST + commit, then call Feishu best- # effort. Previous order (Feishu cancel before local flip) # had a partial-failure window: Feishu succeeds, local commit # crashes → Feishu shows cancelled, Helix shows pending. With # the reversed order, Helix is source of truth; if the Feishu # cancel below fails, the orphaned instance gets reconciled by # the next callback (it sees status=cancelled locally and # no-ops, or by the approval_pending_reaper cron). row.status = DNS_REQ_CANCELLED row.decided_at = datetime.now(UTC) await self.db.flush() audit = AuditService(self.db) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=requester.id, action="cmdb.dns_request.cancel", status="success", resource_type="cmdb_dns_record_request", resource_id=str(row.id), detail={"feishu_instance_code": row.feishu_instance_code}, ) # Commit locally so row + audit are durable BEFORE Feishu. await self.db.commit() self.last_feishu_cancel_synced = True if row.feishu_instance_code: async with best_effort( self.db, label="dns_request_service best-effort step", log=logger, ) as be_1237_step: client = await self._client() # Feishu 撤回 must be addressed as the ORIGINAL applicant, # NOT the acting user. When an admin cancels a leaver's stale # request on their behalf, ``requester`` is the admin, so # resolving ``requester.work_email`` yields the admin's # user_id → cancel_instance is refused and the Feishu instance # stays pending forever (only a desync notice fires). Resolve # the applicant (``row.requester_person_id``) instead. applicant = await self.db.get( IdentityPerson, _uuid.UUID(str(row.requester_person_id)), ) applicant_email = applicant.work_email if applicant else None feishu_user_id = ( await client.resolve_user_id_by_email(applicant_email) if applicant_email else None ) # The approval_code is recorded in the ApprovalRequest's # source_payload — fetch it to avoid duplicating the # flow lookup. approval_request = (await self.db.execute( select(ApprovalRequest).where( ApprovalRequest.external_id == row.feishu_instance_code, ), )).scalar_one_or_none() approval_code = "" if approval_request and approval_request.source_payload: approval_code = str( approval_request.source_payload.get("approval_code") or "", ) if not feishu_user_id: # Couldn't resolve the applicant's Feishu user_id → # can't address the cancel. Route through the desync # branch (mark unsynced + notify) instead of calling # cancel_instance with a null user_id. raise AppException( code="FEISHU_CANCEL_NO_APPLICANT", message=( "could not resolve applicant Feishu user_id for " f"instance {row.feishu_instance_code}" ), status_code=502, ) if approval_code: await client.cancel_instance( instance_code=row.feishu_instance_code, user_id=feishu_user_id, approval_code=approval_code, ) else: # No approval_code on record → the Feishu cancel # endpoint can't be addressed and the instance stays # pending on the Feishu side. Same desync as a # failed call — route it through the except branch # instead of silently reporting synced. raise AppException( code="FEISHU_CANCEL_NO_APPROVAL_CODE", message=( "no approval_code recorded for Feishu " f"instance {row.feishu_instance_code}" ), status_code=502, ) if not be_1237_step.ok: logger.warning( "DNS req %s: best-effort Feishu cancel failed: %s", row.id, be_1237_step.reason(), ) self.last_feishu_cancel_synced = False await notify_feishu_cancel_desync( self.db, flow_label="DNS 解析记录申请", resource_type="cmdb_dns_record_request", resource_id=str(row.id), instance_code=row.feishu_instance_code, # Pass the exception OBJECT, not its rendered text: the # desync notifier does ``isinstance(error, AppException)`` # to pull out the Feishu error code, and a string silently # drops it (caught by the 60009 assertion 2026-08-06). error=be_1237_step.error or RuntimeError("unknown"), trace_id=trace_id, ) await self.db.refresh(row) return await self._to_response(row) # ── Recertify (owner says "still in use") ────────────────── async def recertify_owned( self, *, owner: AuthUser, record_ids: list[str] | None, trace_id: str, ) -> int: """Set ``last_owner_review_at = NOW()`` on every DNS record the caller owns, optionally filtered to ``record_ids``. ``record_ids`` is the list rendered in the Feishu card the owner just clicked; we still re-check ownership server-side because the records might have changed hands between card render and click. Records the caller no longer owns are silently skipped (audit log records the count). """ q = select(CmdbDnsRecord).where( CmdbDnsRecord.owner_person_id == str(owner.id), CmdbDnsRecord.deleted_at.is_(None), ) if record_ids: try: ids = [_uuid.UUID(r) for r in record_ids] except ValueError as exc: raise ValidationError("Invalid record id in payload.") from exc q = q.where(CmdbDnsRecord.id.in_(ids)) rows = (await self.db.execute(q)).scalars().all() now = datetime.now(UTC) count = 0 for r in rows: r.last_owner_review_at = now count += 1 await self.db.flush() audit = AuditService(self.db) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=owner.id, action="cmdb.dns.recertify", status="success", resource_type="cmdb_dns_record", detail={"recertified": count, "scope": "owned"}, ) # Caller owns the transaction — see ``submit`` rationale. await self.db.flush() return count # ── Admin recovery (4-eyes bypass + retry) ───────────────── # # When the server-side 4-eyes guard in handle_feishu_callback fires # (requester == approver), the row lands at ``failed`` with a fixed # error_message fragment ("4-eyes violation"). The standard recovery # path is "fix the Feishu template so the requester isn't routed to # themselves, then submit a fresh request". These two methods cover # the cases where that path is too heavy: # # * ``bypass_execute`` — explicit, audited override. Admin attests # a reason and proceeds to the cloud write WITHOUT the 4-eyes # gate. Every other admin gets a notification so the bypass is # visible. Use for genuine emergencies / single-admin shops where # no peer reviewer exists. # # * ``retry`` — re-open a fresh Feishu instance with the same form # data after the template has been fixed. Cheaper than asking # the user to re-fill the form. Same requester (the row's # ``requester_person_id``), new instance code, ApprovalRequest # row created anew (old one's external_id stays for audit). # # Both methods require ``can_manage_cloud`` (global_admin / # cloud_admin), gate on ``status=failed`` + ``error_message contains # "4-eyes"`` so they cannot be abused on cloud-write failures or # other terminal states, and audit-log every action. _FOUR_EYES_BYPASS_REASON_MIN_LEN = 20 @staticmethod def _is_four_eyes_failure(row: CmdbDnsRecordRequest) -> bool: if row.status != DNS_REQ_FAILED: return False msg = (row.error_message or "").lower() return "4-eyes" in msg or "self-approval" in msg @staticmethod def _is_write_lock_failure(row: CmdbDnsRecordRequest) -> bool: """A row that reached ``failed`` purely because DNS writes were locked at execute time — NOT because the cloud provider call itself blew up. Only these are safe to resume (re-drive the provider write): the approval is valid and no provider mutation was attempted, so there is no double-fire risk. Matches BOTH the current message (``dns_writes_enabled``) and the legacy one (``prod_cloud_writes_enabled`` — rows that failed under the old coupling, before DNS got its own switch), so requests stranded by the old freeze can be resumed after this ships. """ if row.status != DNS_REQ_FAILED: return False msg = (row.error_message or "").lower() return ( "dns_writes_enabled" in msg or "dns writes are locked" in msg or "prod_cloud_writes_enabled" in msg or "cloud writes are globally frozen" in msg ) @staticmethod def _is_stranded_execution_failure(row: CmdbDnsRecordRequest) -> bool: """A row the executing-reaper failed after exhausting its re-arms. Resumable for the same reason a write-lock failure is: the provider was not left half-mutated. ``create`` — the only non-idempotent action — is checked against the live provider on every reaper pass, so reaching this state means the record verifiably is NOT there; every other action is idempotent under a re-drive. """ if row.status != DNS_REQ_FAILED: return False return _STRANDED_EXECUTION_MARKER in (row.error_message or "").lower() async def _list_admin_recipients_excluding( self, exclude_person_id: str, ) -> list[IdentityPerson]: """All active admins with a work_email, minus ``exclude_person_id``. Recipient pool = ``global_admin`` ∪ ``cloud_admin`` ∪ ``it_admin`` — the three roles that can act on cloud DNS. The bypasser themselves is excluded so they don't get a notification about their own action (noise + might mask that the fan-out fired). Same join shape as ``app.scheduled_jobs.failure_notifier._list_admin_recipients``, plus ``cloud_admin`` because cloud_admins are exactly the population a DNS-4-eyes bypass announcement should reach. """ from sqlalchemy import or_ from app.auth.models import AuthRole, AuthUserRole from app.core.rbac import CLOUD_ADMIN_ROLE, GLOBAL_ADMIN_ROLE rows = ( await self.db.execute( select(IdentityPerson) .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) .where( and_( IdentityPerson.deleted_at.is_(None), IdentityPerson.employment_status == "active", IdentityPerson.work_email.isnot(None), IdentityPerson.work_email != "", IdentityPerson.id != exclude_person_id, or_( AuthRole.name == GLOBAL_ADMIN_ROLE, AuthRole.name == CLOUD_ADMIN_ROLE, AuthRole.name == "it_admin", ), ) ) .distinct() ) ).scalars().all() return list(rows) async def _notify_bypass( self, *, row: CmdbDnsRecordRequest, actor: AuthUser, reason: str, ) -> None: """Fire a high-visibility notification to every other admin. Best-effort per recipient — a dispatch failure for one admin does NOT block the cloud write that's about to run. Each failure is logged + lands in the notification outbox for retry. """ from app.notifications.service import ( CardHighlight, LocalizedText, NotificationEnvelope, NotificationService, ) recipients = await self._list_admin_recipients_excluding( exclude_person_id=str(actor.id), ) if not recipients: logger.warning( "DNS 4-eyes bypass on req %s: no other admins to notify " "(actor=%s) — bypass proceeds but no peer visibility", row.id, actor.id, ) return notify = NotificationService(self.db) actor_label = actor.work_email or str(actor.id) fqdn = _display_fqdn(row.name, row.zone_name) title_zh = "⚠️ DNS 申请 4-eyes 拦截被绕过" title_en = "⚠️ DNS request 4-eyes guard bypassed" body_zh = ( f"{actor_label} 显式绕过了 DNS 申请 {fqdn} " f"({row.action}) 的 4-eyes 拦截并执行了变更。\n\n" f"理由:{reason}" ) body_en = ( f"{actor_label} explicitly bypassed the 4-eyes guard on " f"DNS request {fqdn} ({row.action}) and proceeded to " f"execute the change.\n\nReason: {reason}" ) # CardHighlight schema is (label: str, items: list[str] | str) — we # pass single-line values as a bare str so the renderer treats it # as "the only line" (same shape used by lifecycle scope-alert). highlights = [ CardHighlight( label=LocalizedText( zh="申请 ID", en="Request ID", ja="申請 ID", ), items=str(row.id), ), CardHighlight( label=LocalizedText(zh="域名", en="FQDN", ja="FQDN"), items=fqdn, ), CardHighlight( label=LocalizedText(zh="操作", en="Action", ja="操作"), items=row.action, ), CardHighlight( label=LocalizedText(zh="操作人", en="Actor", ja="実行者"), items=actor_label, ), ] for r in recipients: # Shared resolver — see ``app.core.locale``. is_en = locale_is_en(r) try: await notify.dispatch( NotificationEnvelope( recipient=r, title=title_en if is_en else title_zh, body_text=body_en if is_en else body_zh, channels=["feishu", "email"], # Peer-audit notice → OTHER ADMINS, so land # them on the admin DNS surface. Emitted # ``/cloud/dns-requests/{id}`` until # 2026-07-28, which has never been a desktop # route (only the mobile shell has that # page) — so every peer notice 404'd for an # admin reading it on a laptop. The request # identity (FQDN + actor + reason) is in the # body, which is what an audit reader needs. url="/admin/cloud/dns", card_subtitle=LocalizedText( zh="管理员同行审计通告", en="Peer-visible audit notice", ja="管理者間の監査通知", ), card_highlights=highlights, card_highlights_title=LocalizedText( zh="绕过详情", en="Bypass details", ja="バイパスの詳細", ), related_resource_type="cmdb_dns_record_request", related_resource_id=row.id, ), trace_id=f"cmdb.dns_request.bypass:{row.id}", ) except Exception: # noqa: BLE001 logger.exception( "DNS 4-eyes bypass: notification dispatch failed " "req=%s recipient=%s", row.id, r.id, ) async def bypass_4_eyes_execute( self, *, request_id: str, actor: AuthUser, reason: str, trace_id: str, ip_address: str | None = None, user_agent: str | None = None, ) -> DnsRecordRequestResponse: """Admin-attested override: flip the row back to ``approved`` and run the cloud write WITHOUT re-checking 4-eyes. Preconditions enforced: 1. ``actor`` holds ``can_manage_cloud`` (global_admin or cloud_admin). Auditor / management / employee → 403. 2. ``reason`` is at least ``_FOUR_EYES_BYPASS_REASON_MIN_LEN`` characters long. 3. Row exists, status is ``failed``, and the error_message carries the 4-eyes fragment. Bypassing a cloud-write failure would double-fire the provider call. On success: * audit ``cmdb.dns_request.bypassed_4_eyes_execute`` with the reason + original error_message * every OTHER admin gets a Feishu + email notification * row flips to ``approved`` → ``executing`` → ``executed`` (or ``failed`` again with a NEW error_message if the cloud write itself blows up) """ from app.core.rbac import can_manage_cloud if not can_manage_cloud(actor.role_names): raise AppException( code="DNS_REQ_BYPASS_FORBIDDEN", message=( "Only global_admin or cloud_admin can bypass the " "4-eyes guard on a DNS request." ), status_code=403, ) cleaned_reason = (reason or "").strip() if len(cleaned_reason) < self._FOUR_EYES_BYPASS_REASON_MIN_LEN: raise ValidationError( f"Bypass reason must be at least " f"{self._FOUR_EYES_BYPASS_REASON_MIN_LEN} characters — " "this attestation is part of the audit trail." ) # Lock the row. Anything that observes status=failed AND grabs # the lock first wins; concurrent bypass clicks serialize and # the second one fails the precondition check below. row = ( await self.db.execute( select(CmdbDnsRecordRequest) .where(CmdbDnsRecordRequest.id == _uuid.UUID(request_id)) .with_for_update() ) ).scalar_one_or_none() if row is None: raise NotFoundError("DnsRecordRequest", request_id) # Pre-launch P0 fix (Round 45, CR-201): self-bypass is exactly # what 4-eyes exists to prevent. An actor who is both # requester AND a cloud_admin must not be able to skip the # second-pair-of-eyes guard themselves — another admin has to # attest. Without this check, a cloud_admin requests a sensitive # DNS change, the original Feishu approval auto-fails on # self-approval, and the same admin calls the bypass endpoint # with a self-written 20-char reason — defeating the policy. if str(row.requester_person_id) == str(actor.id): raise AppException( code="DNS_REQ_BYPASS_SELF_FORBIDDEN", message=( "You cannot bypass a 4-eyes block on your own DNS " "request — that's exactly what 4-eyes is designed " "to prevent. Another cloud_admin / global_admin " "must execute the bypass." ), status_code=403, ) if not self._is_four_eyes_failure(row): raise AppException( code="DNS_REQ_BYPASS_NOT_APPLICABLE", message=( f"Cannot bypass — request is in status {row.status!r}" f" (need 'failed' with a 4-eyes block). " "Bypass is intentionally limited to 4-eyes failures." ), status_code=409, ) original_error = row.error_message original_instance_code = row.feishu_instance_code or "" # Flip back to APPROVED so ``_execute_approved`` accepts the row. # decided_at is updated to NOW so the timeline reflects the # bypass decision (not the original 4-eyes block). now = datetime.now(UTC) row.status = DNS_REQ_APPROVED row.decided_at = now row.error_message = None await self.db.flush() # Audit BEFORE the cloud write so the bypass attestation is # durable even if the cloud call rolls back the row state. audit = AuditService(self.db) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="cmdb.dns_request.bypassed_4_eyes_execute", status="success", resource_type="cmdb_dns_record_request", resource_id=str(row.id), detail={ "bypass_reason": cleaned_reason, "original_error_message": original_error, "feishu_instance_code": original_instance_code, "action": row.action, "zone_name": row.zone_name, "name": row.name, "record_type": row.record_type, }, ip_address=ip_address, user_agent=user_agent, ) # Fan-out notification. Best-effort — failures land in outbox. await self._notify_bypass(row=row, actor=actor, reason=cleaned_reason) # Commit the approved-marker + audit + outbox rows so the cloud # write below runs in its own transaction (same shape as the # normal Feishu-callback approve path). await self.db.commit() # _execute_approved re-reads the row WITH a FOR UPDATE lock, # re-checks status==APPROVED, then runs the provider write. # The trace_id passed in becomes the audit trace for the # subsequent ``executed`` / ``execute_failed`` row. await self._execute_approved( row, instance_code=f"bypass:{request_id}", ) # Re-read to return the final state (executed / failed). await self.db.refresh(row) final = await self.db.get(CmdbDnsRecordRequest, row.id) if final is None: # Theoretically impossible — we just held a FOR UPDATE lock # on the row. Surface defensively rather than NPE. raise NotFoundError("DnsRecordRequest", request_id) return await self._to_response(final) async def _revalidate_zone_and_provider( self, row: Any, *, action_label: str, ) -> None: """Re-check the two things that can go stale while a request waits. Both admin recovery paths (``retry_after_template_fix`` and ``resume_after_write_lock_cleared``) resurrect a request that may have been parked for days, and a parked request is exactly when these two change: * the admin flips ``allow_dns_request=False`` on the zone — self-service is closed, so resurrecting one must not sneak through on the permission it had at submit time; * DNS hosting migrates (UCloud → Route53) — the stored ``row.integration_id`` is now stale and would route the cloud write to the WRONG tenant. Extracted so ``resume`` inherits them (#2141): it only checked RBAC, the global DNS switch and the lock state before flipping the row to APPROVED and executing, and ``_execute_approved`` re-checks neither. Failing loudly (the requester submits a fresh request) is the designed behaviour — a recovery path must not be more permissive than ``submit``. """ # ``cmdb_domains.integration_id`` represents the REGISTRAR, not # the DNS provider (2026-06-01) — match on ``domain_name`` only; # provider identity is re-resolved separately below. zone_row = await self.db.scalar( select(CmdbDomain).where( CmdbDomain.domain_name == row.zone_name, CmdbDomain.deleted_at.is_(None), ), ) if zone_row is None or not zone_row.allow_dns_request: raise ValidationError( f"Zone '{row.zone_name}' is no longer open for " "self-service DNS requests; ask an admin to re-enable " f"「允许申请 DNS 解析」 before {action_label}." ) # Same resolver ``submit`` uses, so neither recovery path can # become more permissive than the original submission. from app.domains.cmdb.service import CmdbService resolved = await CmdbService.resolve_dns_integration_for_zone( self.db, row.zone_name, ) if resolved is None: raise ValidationError( f"Zone '{row.zone_name}' no longer has a synced " "authoritative DNS provider — IT Admin must seed a " "record from the cloud console first, then ask the " "requester to submit a fresh request." ) resolved_integ_id, resolved_integ_name, _ = resolved if str(row.integration_id) != resolved_integ_id: raise ValidationError( f"Zone '{row.zone_name}' is now authoritatively hosted " f"by '{resolved_integ_name}' (id {resolved_integ_id}); " f"the original request targeted " f"'{row.integration_id}'. DNS hosting moved between " f"submit and {action_label} — the requester must submit " "a fresh request so it routes to the current provider." ) async def retry_after_template_fix( self, *, request_id: str, actor: AuthUser, trace_id: str, ip_address: str | None = None, user_agent: str | None = None, ) -> DnsRecordRequestResponse: from app.core.distributed_lock import LockBusy, distributed_lock try: async with distributed_lock( f"cmdb_dns_retry:{request_id}", ttl=120, ): return await self._retry_after_template_fix_locked( request_id=request_id, actor=actor, trace_id=trace_id, ip_address=ip_address, user_agent=user_agent, ) except LockBusy as exc: raise AppException( code="DNS_REQ_RETRY_IN_FLIGHT", message=( "This DNS request is already being retried. " "Please retry in a few seconds." ), status_code=409, ) from exc async def _retry_after_template_fix_locked( self, *, request_id: str, actor: AuthUser, trace_id: str, ip_address: str | None = None, user_agent: str | None = None, ) -> DnsRecordRequestResponse: """Open a fresh Feishu approval instance for a row that's stuck at ``failed`` with a 4-eyes block. Use case: admin fixed the Feishu template's approver routing (so the requester is no longer in any approver node) and now wants to push the same request through without making the user re-fill the form. Same preconditions as ``bypass_4_eyes_execute`` (admin role, 4-eyes failure) but the resulting flow goes through Feishu as normal — no bypass, the new template-side approvers must still approve, and the server-side 4-eyes guard runs again on the new callback. """ from app.core.rbac import can_manage_cloud if not can_manage_cloud(actor.role_names): raise AppException( code="DNS_REQ_RETRY_FORBIDDEN", message=( "Only global_admin or cloud_admin can retry a " "blocked DNS request." ), status_code=403, ) row = ( await self.db.execute( select(CmdbDnsRecordRequest) .where(CmdbDnsRecordRequest.id == _uuid.UUID(request_id)) .with_for_update() ) ).scalar_one_or_none() if row is None: raise NotFoundError("DnsRecordRequest", request_id) if not self._is_four_eyes_failure(row): raise AppException( code="DNS_REQ_RETRY_NOT_APPLICABLE", message=( f"Cannot retry — request is in status {row.status!r}" f" (need 'failed' with a 4-eyes block). " "Retry is intentionally limited to 4-eyes failures " "to avoid double-firing cloud writes." ), status_code=409, ) await self._revalidate_zone_and_provider(row, action_label="retry") # Resolve the original requester so the new Feishu instance is # opened under their identity (not the admin clicking retry — # they're just a recovery operator). requester = await self.db.get( IdentityPerson, _uuid.UUID(str(row.requester_person_id)), ) if ( requester is None or not requester.work_email or requester.employment_status != "active" ): raise AppException( code="DNS_REQ_RETRY_REQUESTER_INACTIVE", message=( "Original requester is no longer active / has no " "work_email; cannot reopen the Feishu instance under " "their identity. Ask them to submit a fresh request." ), status_code=409, ) # Re-build the Feishu form from the row fields. Same shape as # ``submit`` — we don't refactor to share since the validation # surface differs (submit re-validates user input; retry trusts # the durable row). flow = await get_approval_flow(_FEISHU_FLOW_KEY, session=self.db) approval_code = flow.external_code client = await self._client() definition = await client.get_approval_definition(approval_code) widgets = _parse_template_widgets(definition.get("form", "[]")) action_label = { DNS_REQ_ACTION_CREATE: "新增 / Create", DNS_REQ_ACTION_UPDATE: "修改 / Update", DNS_REQ_ACTION_DELETE: "删除 / Delete", DNS_REQ_ACTION_ENABLE: "启用 / Enable", DNS_REQ_ACTION_DISABLE: "停用 / Disable", }[row.action] field_values: dict[str, Any] = { "action": action_label, "fqdn": _display_fqdn(row.name, row.zone_name), "zone": row.zone_name, "name": row.name, "record_type": row.record_type, "value": row.value, "ttl": str(row.ttl), "purpose": row.purpose, } form_payload = _build_dns_form_payload(widgets, field_values) feishu_user_id = await client.resolve_user_id_by_email( requester.work_email, ) semantic_payload = { "request_id": str(row.id), "action": row.action, "integration_id": str(row.integration_id), "zone_name": row.zone_name, "name": row.name, "record_type": row.record_type, "value": row.value, "ttl": row.ttl, "purpose": row.purpose, } intent = await submit_with_durable_intent( self.db, client=client, scope=f"cmdb_dns_retry:{row.id}", applicant_id=requester.id, semantic_payload=semantic_payload, provider_payload={ "approval_code": approval_code, "identity_type": "user_id", "identity_value": feishu_user_id, "form": form_payload, }, finalization_payload={ **semantic_payload, "mode": "retry", "approval_request_id": str(_uuid.uuid4()), "approval_code": approval_code, "actor_user_id": str(actor.id), }, trace_id=trace_id, ) row = await self._finalize_submission_intent( intent.id, recovered=intent.attempt_count > 1, ip_address=ip_address, user_agent=user_agent, ) return await self._to_response(row) async def resume_after_write_lock_cleared( self, *, request_id: str, actor: AuthUser, trace_id: str, ip_address: str | None = None, user_agent: str | None = None, ) -> DnsRecordRequestResponse: """Re-drive a DNS request that reached ``failed`` without the provider ever being mutated. Two such failures qualify: * DNS writes were locked at execute time (``cmdb.dns_writes_enabled`` / the legacy prod-cloud freeze) — no provider call was attempted at all; * execution was repeatedly interrupted and the reaper gave up (:meth:`_is_stranded_execution_failure`) — ``create`` is verified absent on the provider each reaper pass, and every other action is idempotent. Either way the Feishu approval already succeeded, so resuming just re-runs the cloud write on the same, still-valid approval — no new approval instance, no bypass. Preconditions: 1. ``actor`` holds ``can_manage_cloud`` (global_admin / cloud_admin). Others → 403. 2. Row exists, ``status=failed`` with one of the two markers above. A genuine cloud-write failure is NOT resumable here — that path may have already mutated the provider and re-driving it could double-fire. 3. DNS writes are currently ENABLED. Resuming into a still- locked switch would just re-fail the row; refuse up front with a clear message so the admin flips the switch first. On success the row runs ``approved → executing → executed`` (or ``failed`` again with a fresh error if the provider write itself blows up), and a confirmation comment is posted back to the original Feishu approval instance. """ from app.core.rbac import can_manage_cloud from app.domains.cmdb.prod_write_lock import dns_writes_enabled if not can_manage_cloud(actor.role_names): raise AppException( code="DNS_REQ_RESUME_FORBIDDEN", message=( "Only global_admin or cloud_admin can resume a " "write-locked DNS request." ), status_code=403, ) # Refuse to resume while the switch is still locked — otherwise # ``_execute_approved`` would just re-fail the row and post a # second block comment. Check BEFORE taking the row lock so a # locked switch fails fast without touching the row. if not await dns_writes_enabled(self.db): raise AppException( code="DNS_REQ_RESUME_STILL_LOCKED", message=( "DNS writes are still locked " "(cmdb.dns_writes_enabled=false). Re-enable DNS writes " "in 平台设置 before resuming this request." ), status_code=409, ) # Lock the row so concurrent resume clicks / a resume racing the # original callback serialize; the loser fails the precondition. row = ( await self.db.execute( select(CmdbDnsRecordRequest) .where(CmdbDnsRecordRequest.id == _uuid.UUID(request_id)) .with_for_update() ) ).scalar_one_or_none() if row is None: raise NotFoundError("DnsRecordRequest", request_id) if not ( self._is_write_lock_failure(row) or self._is_stranded_execution_failure(row) ): raise AppException( code="DNS_REQ_RESUME_NOT_APPLICABLE", message=( f"Cannot resume — request is in status {row.status!r} " "(need 'failed' due to a DNS write-lock or an " "interrupted execution). Resume is limited to failures " "where the provider was verifiably not mutated; a real " "cloud-write failure must be re-submitted as a fresh " "request." ), status_code=409, ) # A write-lock hold can last DAYS, and those are exactly the days # in which an admin closes the zone to self-service or DNS # hosting migrates. ``_execute_approved`` re-checks neither, and # ``_execute_create/update/delete`` hit the provider stored on # the ROW — so a resume used to execute against the submit-time # permission and the submit-time tenant (#2141). The retry path # has had both gates since it was written; resume never got them. # Raising sends the requester to a fresh request, which is the # right outcome: a recovery path must not be more permissive # than ``submit``. await self._revalidate_zone_and_provider(row, action_label="resume") original_error = row.error_message original_instance_code = row.feishu_instance_code or "" # Flip back to APPROVED so ``_execute_approved`` accepts the row. # decided_at is left untouched — the original approval decision is # still the real one; only the deferred write is being resumed. row.status = DNS_REQ_APPROVED row.error_message = None await self.db.flush() audit = AuditService(self.db) await audit.log( trace_id=trace_id, actor_type="user", actor_user_id=actor.id, action="cmdb.dns_request.resumed_after_write_lock", status="success", resource_type="cmdb_dns_record_request", resource_id=str(row.id), detail={ "original_error_message": original_error, "feishu_instance_code": original_instance_code, "action": row.action, "zone_name": row.zone_name, "name": row.name, "record_type": row.record_type, }, ip_address=ip_address, user_agent=user_agent, ) # Commit the approved-marker + audit so the cloud write below runs # in its own transaction (same shape as the Feishu-callback path). await self.db.commit() # Re-drive the provider write. ``_execute_approved`` re-locks the # row, re-checks status==APPROVED and the (now-open) DNS switch, # then hits the provider. On failure it posts its own block # comment; on success it is silent, so we post the confirmation. exec_instance_code = original_instance_code or f"resume:{request_id}" # Read the PK BEFORE the call: the fail-path rolls back, which expires # every ORM attribute, so ``row.id`` afterwards is a lazy load outside # greenlet context (``MissingGreenlet``) — the same trap the 2026-08-14 # incident sprang. Never touch ``row`` after ``_execute_approved``. resumed_row_id = row.id await self._execute_approved(row, instance_code=exec_instance_code) final = await self.db.get(CmdbDnsRecordRequest, resumed_row_id) if final is None: raise NotFoundError("DnsRecordRequest", request_id) # ``_execute_approved`` committed, expiring the identity-map row; # refresh before serializing so ``_to_response`` reads live # columns instead of triggering a lazy load. Mirrors the refresh # in ``bypass_4_eyes_execute``. await self.db.refresh(final) if final.status == DNS_REQ_EXECUTED and original_instance_code: await self._post_blocked_feishu_comment( instance_code=original_instance_code, requester_person_id=final.requester_person_id, comment_text=( "✅ 此 DNS 申请已由 IT 管理员重跑并成功执行，DNS 记录已" "按审批内容变更完成。\n" "✅ This DNS request was resumed by an IT admin and " "executed successfully — the DNS record has been changed " "as approved." ), ) return await self._to_response(final) # ── Feishu callback (executed from approval/service.py) ───── async def handle_feishu_callback( self, *, approval_request: ApprovalRequest, instance_detail: dict[str, Any], ) -> None: """Called by ApprovalService.handle_feishu_callback after it has acquired the row-level lock on the ApprovalRequest and confirmed the request hasn't been settled yet. We re-load the DNS request row and dispatch on ``status`` of the Feishu instance. The actual provider write is delegated to the same helpers the admin /admin/cloud/dns page already uses, so the bytes hitting Tencent / Route53 are identical. """ feishu_status = str(instance_detail.get("status") or "").upper() # Map Feishu status to our state machine. PENDING is a no-op # callback (Feishu sometimes redelivers). if feishu_status in ("PENDING", "RECALL"): return # ApprovalRequest.source_payload was set at submit time. payload = approval_request.source_payload or {} dns_req_id = payload.get("dns_request_id") if not dns_req_id: logger.warning( "Feishu DNS callback %s missing dns_request_id in source_payload", approval_request.external_id, ) return row = await self.db.get( CmdbDnsRecordRequest, _uuid.UUID(str(dns_req_id)), ) if row is None: logger.warning( "Feishu DNS callback %s references missing request %s", approval_request.external_id, dns_req_id, ) return if row.status in DNS_REQ_TERMINAL_STATES: logger.info( "DNS request %s already in terminal state %s — skipping", row.id, row.status, ) return now = datetime.now(UTC) if feishu_status == "REJECTED": row.status = DNS_REQ_REJECTED row.decided_at = now await self.db.flush() await self._audit( trace_id=f"feishu:{approval_request.external_id}", action="cmdb.dns_request.rejected", row=row, ) return if feishu_status == "CANCELED": row.status = DNS_REQ_CANCELLED row.decided_at = now await self.db.flush() await self._audit( trace_id=f"feishu:{approval_request.external_id}", action="cmdb.dns_request.cancelled", row=row, ) return if feishu_status != "APPROVED": logger.warning( "Feishu DNS callback %s unknown status %s — leaving pending", approval_request.external_id, feishu_status, ) return # APPROVED — guard against self-approval (4-eyes principle). # The Feishu approval template SHOULD route DNS requests away # from the requester, but a misconfigured template (or an admin # editing the flow node list) could let cloud_admin / it_admin # approve their own request. We enforce this server-side here # so a future template misconfig cannot weaken the 4-eyes # invariant. instance_detail.task_list shape is the same one # ApprovalService._build_admin_timeline already consumes # (verified against prod approval instances 2026-05-15). if await self._is_self_approval( instance_detail=instance_detail, requester_person_id=str(row.requester_person_id), ): row.status = DNS_REQ_FAILED row.decided_at = now row.error_message = ( "Blocked: requester appears as an approver on this " "approval (4-eyes violation). Re-route the Feishu " "approval template so the requester is not in any " "approver node." ) await self.db.flush() await self._audit( trace_id=f"feishu:{approval_request.external_id}", action="cmdb.dns_request.execute_blocked_self_approval", row=row, ) logger.warning( "DNS request %s blocked: self-approval detected " "(requester_person_id=%s feishu_instance=%s)", row.id, row.requester_person_id, approval_request.external_id, ) await self._post_blocked_feishu_comment( instance_code=approval_request.external_id, requester_person_id=row.requester_person_id, comment_text=( "⚠️ 此 DNS 申请虽已审批通过，但 Helix 出于安全（四眼原则）" "未执行：申请人本人同时是本审批流的审批人之一，系统禁止" "「自己审批自己的申请」。请将飞书审批模板中的审批人节点改为" "他人后，重新发起本申请。\n" "⚠️ This DNS request was approved but Helix did NOT apply it " "for security (4-eyes principle): the requester is also one of " "the approvers on this flow, and self-approval is forbidden. " "Re-route the Feishu approval template so the approver is " "someone other than the requester, then submit a new request." ), ) return # APPROVED — execute. Mark approved + executing, commit so # we never lose the approval if the cloud write blows up # mid-flight. row.status = DNS_REQ_APPROVED row.decided_at = now await self.db.flush() await self._audit( trace_id=f"feishu:{approval_request.external_id}", action="cmdb.dns_request.approved", row=row, ) # Commit the approved-marker and STOP. The cloud write is not run # inline any more — ``cmdb.dns_request_executor`` picks the row up # within a minute and runs it in the worker. # # Why (2026-08-14): this method runs inside the Feishu webhook's # HTTP request. A DNSPod / Route53 write takes seconds, and Feishu # gives a webhook ~3s before it declares failure and redelivers — # instance 0778DA79 was delivered four times in 26 seconds while # the first delivery was still inside the cloud call. Every extra # delivery is another callback racing the row's state machine, and # any transport hiccup on the slow one (client disconnect, deploy # restart, request cancellation) tears the handler down mid-write # with no exception path to mark the row. # # ``approved`` is the durable handoff point the state machine was # already designed around, so the executor cron needs no new queue # table. Admin-initiated paths (``bypass_execute`` / # ``resume_after_write_lock_cleared``) keep calling # :meth:`_execute_approved` inline — a human is waiting on that # response and nothing is redelivering it. await self.db.commit() async def _is_self_approval( self, *, instance_detail: dict[str, Any], requester_person_id: str, ) -> bool: """Return True if the requester's Feishu open_id appears as an APPROVED approver in ``instance_detail.task_list``. Algorithm: 1. Load the requester's ``work_email`` from IdentityPerson. If they no longer have an active email (account deleted, email cleared) we cannot verify → return True to fail closed; the request is rejected and audit logged. 2. Resolve work_email → Feishu open_id via the cached ``FeishuApprovalClient.resolve_open_id_by_email``. A resolve failure means Feishu has no user for this email → also fail closed. 3. Walk ``task_list`` looking for any entry whose ``status`` ∈ {"APPROVED"} AND whose ``open_id`` matches the requester's. Status keys ``"PENDING"`` / ``"REJECTED"`` / ``"TRANSFERRED"`` / ``"CANCELED"`` do not count — only APPROVED actions can complete an approval. We DELIBERATELY use ``open_id`` rather than ``user_id`` because Feishu's approval ``task_list`` payload always carries both, and our submit path resolves work_email→user_id to OPEN the instance — so the requester's open_id is the stable identifier across the lifecycle. """ from app.identity.person_models import IdentityPerson task_list = instance_detail.get("task_list") if not isinstance(task_list, list) or not task_list: # FAIL CLOSED: a genuine four-eyes approval always yields a # task_list entry naming the approver. An APPROVED instance # with a missing/stripped/empty task_list (Feishu payload-shape # drift, a redelivery variant, or an auto-approve node) cannot # be verified as non-self-approval, so we must treat it AS a # self-approval and block — every other uncertainty branch in # this function already fails closed. (Was fail-OPEN, letting a # high-blast DNS write execute without the four-eyes check.) return True person = await self.db.get( IdentityPerson, _uuid.UUID(requester_person_id), ) if person is None or not person.work_email: logger.warning( "4-eyes check fail-closed: requester %s has no " "work_email; cannot resolve to Feishu open_id", requester_person_id, ) return True try: client = await self._client() requester_open_id = await client.resolve_open_id_by_email( person.work_email, ) except Exception as exc: # noqa: BLE001 logger.warning( "4-eyes check fail-closed: Feishu open_id resolve " "failed for requester %s (%s): %s", requester_person_id, person.work_email, exc, ) return True # Bound task_list scan to first 500 entries (mirrors the cap # in ApprovalService._build_admin_timeline so a malicious / # malformed payload cannot DoS this check). for task in task_list[:500]: if not isinstance(task, dict): continue status = str(task.get("status") or "").upper() if status != "APPROVED": continue task_open_id = task.get("open_id") if isinstance(task_open_id, str) and task_open_id == requester_open_id: return True return False async def _post_blocked_feishu_comment( self, *, instance_code: str, requester_person_id: Any, comment_text: str, ) -> None: """Best-effort: write an explanatory comment back onto the Feishu approval instance when a *fully-approved* DNS request will NOT be applied. These paths (4-eyes self-approval block, inactive-requester block, cloud-write failure) all leave the Feishu card reading 已通过 while the DNS record is untouched. The only local trace is ``cmdb_dns_record_requests.error_message`` — which the requester has no reason to open. Surfacing the reason on the approval card itself means the requester finds out in Feishu instead of opening a support investigation every time. Strictly best-effort: a Feishu hiccup here must never derail the block/fail bookkeeping the caller already did. The comment is authored as the requester (the only user_id we can reliably resolve — and the same identity the 4-eyes check already matched). All errors are swallowed + logged. """ try: from app.identity.person_models import IdentityPerson requester = await self.db.get( IdentityPerson, requester_person_id, ) if requester is None or not requester.work_email: logger.warning( "DNS request: skipping Feishu block-comment — " "requester %s has no work_email", requester_person_id, ) return client = await self._client() user_id = await client.resolve_user_id_by_email( requester.work_email, ) await client.add_instance_comment( instance_code=instance_code, user_id=user_id, content_text=comment_text, ) except Exception as exc: # noqa: BLE001 logger.warning( "DNS request: failed to post Feishu block-comment on " "instance %s: %s", instance_code, exc, ) # ── Out-of-band execution (cron-driven) ───────────────────── # # The Feishu callback no longer performs the cloud write (see # handle_feishu_callback). It commits ``approved`` and returns; these # two methods, driven by ``cmdb.dns_request_executor`` and # ``cmdb.dns_request_execution_reaper``, own everything after that: # # approved ──executor──▶ executing ──▶ executed / failed # ▲ │ # └──────reaper────────────┘ (re-arm, bounded by attempts) # # The reaper exists because ``executing`` is the one state with no # owner: whatever process flipped the row into it is, by definition, # gone. Before 2026-08-14 nothing swept it and a single interrupted # write parked a request at 处理中 permanently. async def execute_approved_backlog( self, *, batch_limit: int = 50, ) -> dict[str, int]: """Run the cloud write for every request sitting at ``approved``. Per-row isolated: one row's blow-up must not abandon the rest of the backlog. ``_execute_approved`` already lands provider errors on the row as ``failed``; the guard here is for everything that can fail outside it (session wedged, row vanished, provider credentials undecryptable). """ rows = ( await self.db.execute( select(CmdbDnsRecordRequest) .where(CmdbDnsRecordRequest.status == DNS_REQ_APPROVED) .order_by(CmdbDnsRecordRequest.decided_at.asc()) .limit(batch_limit), ) ).scalars().all() executed = failed = errors = 0 for row in rows: row_id = row.id try: await self._execute_approved( row, row.feishu_instance_code or "", ) except Exception: # noqa: BLE001 errors += 1 logger.exception( "dns_request_executor: row %s raised out of " "_execute_approved", row_id, ) try: await self.db.rollback() except Exception: # noqa: BLE001 logger.exception( "dns_request_executor: rollback failed after row %s", row_id, ) continue settled = await self.db.get(CmdbDnsRecordRequest, row_id) if settled is not None and settled.status == DNS_REQ_EXECUTED: executed += 1 elif settled is not None and settled.status == DNS_REQ_FAILED: failed += 1 return { "checked": len(rows), "executed": executed, "failed": failed, "errors": errors, } async def reconcile_stranded_executions( self, *, stale_minutes: int = 10, max_attempts: int = 3, batch_limit: int = 50, ) -> dict[str, int]: """Un-strand requests parked at ``executing``. A row only reaches ``executing`` from a process that intended to finish it, so anything still there after ``stale_minutes`` lost its owner (interrupted request, container replaced mid-deploy, a fail-path that could not write). ``stale_minutes`` must stay well above the provider SDK's own timeout (30s) so a genuinely in-flight write is never yanked out from under itself. ``create`` is the one non-idempotent action, so it is reconciled against the provider before anything else: if the record is already there the write DID land and only Helix's bookkeeping is missing — finish it rather than re-driving and creating a duplicate. Every other action is idempotent (re-setting a value, deleting a record already gone), so it is simply re-armed to ``approved`` for the executor. Rows that burn through ``max_attempts`` land in ``failed`` with a marker :meth:`_is_stranded_execution_failure` recognises, so an admin can resume them from the UI instead of asking the requester to walk the whole approval chain again. """ cutoff = datetime.now(UTC) - timedelta(minutes=stale_minutes) rows = ( await self.db.execute( select(CmdbDnsRecordRequest) .where( and_( CmdbDnsRecordRequest.status == DNS_REQ_EXECUTING, CmdbDnsRecordRequest.updated_at < cutoff, ), ) .order_by(CmdbDnsRecordRequest.updated_at.asc()) .limit(batch_limit), ) ).scalars().all() healed = rearmed = exhausted = errors = 0 # One provider listing per integration, not per row. cloud_cache: dict[str, list[Any]] = {} for row in rows: row_id = row.id try: landed_pk: str | None = None if row.action == DNS_REQ_ACTION_CREATE: records = cloud_cache.get(str(row.integration_id)) if records is None: provider = await _load_dns_provider( self.db, row.integration_id, ) try: records = list(await provider.list_dns_records()) finally: await provider.close() cloud_cache[str(row.integration_id)] = records match = next( ( r for r in records if r.zone_name == row.zone_name and r.name == row.name and r.record_type == row.record_type and r.value == row.value ), None, ) if match is not None: landed_pk = await self._finalize_create( row, match.record_id, ) if landed_pk is not None: row.status = DNS_REQ_EXECUTED row.executed_at = datetime.now(UTC) row.executed_record_pk = landed_pk await self.db.flush() await self._audit( trace_id=f"reaper:{row.feishu_instance_code or row_id}", action="cmdb.dns_request.execute_reconciled", row=row, extra={ "executed_record_pk": landed_pk, "note": ( "cloud record already present — the " "interrupted write had landed" ), }, ) await self.db.commit() healed += 1 continue if (row.execution_attempts or 0) >= max_attempts: row.status = DNS_REQ_FAILED row.error_message = STRANDED_EXECUTION_ERROR row.executed_at = datetime.now(UTC) await self.db.flush() await self._audit( trace_id=f"reaper:{row.feishu_instance_code or row_id}", action="cmdb.dns_request.execute_stranded", row=row, extra={"attempts": row.execution_attempts}, ) await self.db.commit() exhausted += 1 await self._post_blocked_feishu_comment( instance_code=row.feishu_instance_code or "", requester_person_id=row.requester_person_id, comment_text=( "⚠️ 此 DNS 申请已审批通过，但多次执行都被中断，DNS " "记录未改动。审批仍然有效，IT 管理员点「重跑」即可" "执行，无需重新发起。\n" "⚠️ This DNS request was approved but execution " "was interrupted repeatedly; the DNS record was " "NOT changed. The approval is still valid — an IT " "admin can resume it; no need to re-submit." ), ) continue row.status = DNS_REQ_APPROVED await self.db.flush() await self._audit( trace_id=f"reaper:{row.feishu_instance_code or row_id}", action="cmdb.dns_request.execute_rearmed", row=row, extra={"attempts": row.execution_attempts}, ) await self.db.commit() rearmed += 1 except Exception: # noqa: BLE001 errors += 1 logger.exception( "dns_request_execution_reaper: row %s failed to " "reconcile", row_id, ) try: await self.db.rollback() except Exception: # noqa: BLE001 logger.exception( "dns_request_execution_reaper: rollback failed " "after row %s", row_id, ) return { "checked": len(rows), "healed": healed, "rearmed": rearmed, "exhausted": exhausted, "errors": errors, } async def _execute_approved( self, row: CmdbDnsRecordRequest, instance_code: str, ) -> None: """Run the actual cloud DNS mutation. Called only after the approval marker has been committed so a cloud-write failure leaves the row at ``approved`` (not lost) → ``failed``. Concurrency guard: Feishu webhooks ARE retried on 5xx (and a manual click + webhook race is possible). Take a SELECT FOR UPDATE row lock and re-check ``status`` before flipping it to EXECUTING. Without this lock two concurrent calls both pass the ``status in (APPROVED, EXECUTING)`` guard, both flip the row, and both call ``provider.create_dns_record`` / ``delete_dns_record`` — double-creating the record (or double-deleting one that another tenant has reused). """ # Re-read the row WITH a row-level lock so concurrent webhook # callbacks serialize. The lock is released when the surrounding # transaction commits below. locked = await self.db.execute( select(CmdbDnsRecordRequest) .where(CmdbDnsRecordRequest.id == row.id) .with_for_update(), ) row = locked.scalar_one_or_none() if row is None: return # ``EXECUTING`` is short-circuited here so a redelivery that # arrives between our flip-to-EXECUTING and the cloud-write does # NOT also fire the cloud call. The original handler retains # ownership and either lands EXECUTED or FAILED below. if row.status != DNS_REQ_APPROVED: logger.info( "DNS request %s: skipping execute — status=%s (already " "executing/executed/failed elsewhere)", row.id, row.status, ) return # Re-check requester employment_status. Between Feishu submit # and the approval landing, the requester may have entered # leaver_pending — executing the DNS write at that point would # land a row owned by someone who's mid-offboarding and the # T+0 force-transfer scan has already walked past their assets. # The submit-side resolve_owner_input guard catches outbound # owner-transfers but not the leaver-as-requester case, so # gate it here at the durability boundary. from app.identity.person_models import IdentityPerson requester = await self.db.get( IdentityPerson, row.requester_person_id, ) if requester is None or requester.employment_status != "active": logger.warning( "DNS request %s: requester %s is no longer active " "(status=%s) — failing the request instead of executing", row.id, row.requester_person_id, requester.employment_status if requester else "deleted", ) row.status = DNS_REQ_FAILED row.error_message = ( "Requester is no longer an active employee; " "refusing to execute." ) row.executed_at = datetime.now(UTC) await self.db.flush() await self._audit( trace_id=f"feishu:{instance_code}", action="cmdb.dns_request.execute_blocked_inactive_requester", row=row, extra={ "requester_employment_status": ( requester.employment_status if requester else "deleted" ), }, ) await self.db.commit() await self._post_blocked_feishu_comment( instance_code=instance_code, requester_person_id=row.requester_person_id, comment_text=( "⚠️ 此 DNS 申请虽已审批通过，但 Helix 未执行：申请人在审批" "期间已离职/转为非在职状态，系统拒绝以其名义改动 DNS。如仍" "需变更，请由在职同事重新发起。\n" "⚠️ This DNS request was approved but Helix did NOT apply it: " "the requester is no longer an active employee, so the change " "was refused. If still needed, have an active employee submit " "a new request." ), ) return # Honour the DNS-specific write control (``cmdb.dns_writes_enabled``, # 2026-07-06). This USED to gate on ``cmdb.prod_cloud_writes_enabled`` # — the prod-IAM kill-switch — which coupled every DNS request to an # unrelated incident freeze: an IAM lockdown stalled and failed # already-approved DNS changes even though a DNS record edit is low # blast-radius and trivially reversible. DNS now has its own switch, # defaulting to ALLOWED, so a normal environment executes and only an # explicit DNS freeze blocks here. See ``prod_write_lock.dns_writes``. # # When it IS locked: fail the row (rather than silently parking it at # ``approved``). There is NO reconciler that re-drives ``approved`` DNS # rows and the Feishu callback returns 200 so it is never redelivered — # a parked row would strand invisibly. Marking it ``failed`` with a # clear reason matches the sibling non-executed paths above # (self-approval / inactive requester); unlike the old "submit a fresh # request" model, an admin can now click 「重跑」 (resume) once the # switch is flipped back — the approval is still valid, only the write # was deferred. The Feishu comment tells the requester exactly that. from app.domains.cmdb.prod_write_lock import ( DNS_LOCKED_MESSAGE, dns_writes_enabled, ) if not await dns_writes_enabled(self.db): logger.warning( "DNS request %s: DNS writes are locked " "(cmdb.dns_writes_enabled=false) — failing the request; an " "admin can resume it once writes are re-enabled", row.id, ) row.status = DNS_REQ_FAILED row.error_message = DNS_LOCKED_MESSAGE await self.db.flush() await self._audit( trace_id=f"feishu:{instance_code}", action="cmdb.dns_request.execute_deferred_dns_lock", row=row, ) await self.db.commit() await self._post_blocked_feishu_comment( instance_code=instance_code, requester_person_id=row.requester_person_id, comment_text=( "⚠️ 此 DNS 申请虽已审批通过，但 Helix 未执行：DNS 写操作" "当前被管控开关锁定（cmdb.dns_writes_enabled=false）。审批" "仍然有效，IT 管理员解除锁定后点「重跑」即可执行，无需重新" "发起。\n" "⚠️ This DNS request was approved but Helix did NOT apply " "it: DNS writes are currently locked " "(cmdb.dns_writes_enabled=false). The approval is still " "valid — an IT admin can resume it once DNS writes are " "re-enabled; no need to re-submit." ), ) return row.status = DNS_REQ_EXECUTING row.execution_attempts = (row.execution_attempts or 0) + 1 await self.db.flush() await self.db.commit() # Snapshot the identifiers as plain values BEFORE the cloud write. # ``db.rollback()`` in the fail-path expires every ORM attribute on # ``row``, so any later ``row.id`` / ``row.requester_person_id`` is # a lazy refresh — which raises ``MissingGreenlet`` outside an # awaited context and kills the fail-path itself. That is exactly # how request 85c908bf survived a rejected DNSPod write stranded at # EXECUTING with no ``failed`` state, no audit row, and no Feishu # comment (2026-08-14). Never touch ``row`` after the rollback. row_id = row.id requester_person_id = row.requester_person_id action = row.action try: if action == DNS_REQ_ACTION_CREATE: executed_pk = await self._execute_create(row) elif action == DNS_REQ_ACTION_UPDATE: executed_pk = await self._execute_update(row) elif action == DNS_REQ_ACTION_DELETE: executed_pk = await self._execute_delete(row) elif action == DNS_REQ_ACTION_ENABLE: executed_pk = await self._execute_set_status(row, enable=True) elif action == DNS_REQ_ACTION_DISABLE: executed_pk = await self._execute_set_status(row, enable=False) else: raise AppException( code="DNS_REQ_BAD_ACTION", message=f"Unknown action {action!r}", status_code=500, ) except Exception as exc: # noqa: BLE001 logger.exception( "DNS request %s failed during cloud write", row_id, ) await self._mark_execution_failed( row_id=row_id, requester_person_id=requester_person_id, instance_code=instance_code, exc=exc, ) return # Success — mark executed. row = await self.db.get(CmdbDnsRecordRequest, row_id) if row is not None: row.status = DNS_REQ_EXECUTED row.executed_at = datetime.now(UTC) row.executed_record_pk = executed_pk await self.db.flush() await self._audit( trace_id=f"feishu:{instance_code}", action="cmdb.dns_request.executed", row=row, extra={"executed_record_pk": executed_pk}, ) await self.db.commit() async def _mark_execution_failed( self, *, row_id: _uuid.UUID, requester_person_id: str, instance_code: str, exc: BaseException, ) -> None: """Land a cloud-write failure on the row as ``failed`` + audit. Takes only plain values — no ORM instance. The caller's ``row`` is expired by the rollback below, and re-reading any attribute off it would raise ``MissingGreenlet`` and abort this handler, leaving the request stranded at EXECUTING forever (see :meth:`_execute_approved`). Every step is individually guarded: the whole point of this method is that it runs when the session is already in a bad state, so its own failure must be loud (a log line an operator can find) rather than silent (an exception escaping back into a stuck row). If the state write itself cannot land, the executing-reaper cron is the backstop that reconciles the row against the provider. """ try: await self.db.rollback() except Exception: # noqa: BLE001 logger.exception( "DNS request %s: rollback failed in fail-path; " "session may be unusable", row_id, ) # Round-28 hardening: the raw provider exception (Route53 / Aliyun # / DNSPod) often embeds AccessKeyId fragments, ARNs, account IDs, # region names, signed-URL pieces — anything that would leak # provider topology to the requester through # ``GET /dns-record-requests/{id}``. Surface only a generic message # on the row (the requester sees this); keep the redacted # ``str(exc)`` in the admin-only audit detail. public_msg = _public_error_message(exc) try: row = await self.db.get(CmdbDnsRecordRequest, row_id) if row is None: return row.status = DNS_REQ_FAILED row.error_message = public_msg row.executed_at = datetime.now(UTC) await self.db.flush() from app.domains.cmdb.providers.exceptions import ( redact_provider_error, ) await self._audit( trace_id=f"feishu:{instance_code}", action="cmdb.dns_request.execute_failed", row=row, extra={"error": redact_provider_error(exc, max_len=500)}, ) await self.db.commit() except Exception: # noqa: BLE001 logger.exception( "DNS request %s: could NOT record the execution failure — " "the row is left at executing for the reaper to reconcile " "(original error: %s)", row_id, type(exc).__name__, ) # The block commits, so a SAVEPOINT is the wrong primitive (see # app/core/txn). Hand the caller a usable session instead: a failed # flush here deactivates the transaction, and this method returns # NORMALLY, so the poison surfaces in the caller's next statement — # ``self.db.get`` in the executor cron's batch loop (outside its # try, so one bad row skips every remaining approved request) and # ``self.db.refresh`` on the admin retry path (a 500 on a request # that merely failed to execute). try: await self.db.rollback() except Exception: # noqa: BLE001 logger.exception( "DNS request %s: rollback failed after the fail-path " "write; session may be unusable", row_id, ) return await self._post_blocked_feishu_comment( instance_code=instance_code, requester_person_id=requester_person_id, comment_text=( "⚠️ 此 DNS 申请已审批通过，但执行失败，DNS 记录未改动：" f"{public_msg} 请联系 IT 管理员处理，或稍后重新发起。\n" "⚠️ This DNS request was approved but execution FAILED; " "the DNS record was NOT changed: " f"{public_msg} Please contact your IT admin or retry " "by submitting a new request." ), ) async def _execute_create( self, row: CmdbDnsRecordRequest, ) -> str | None: provider = await _load_dns_provider(self.db, row.integration_id) try: dns_write = DnsRecordWrite( zone_name=row.zone_name, name=row.name, record_type=row.record_type, value=row.value, ttl=row.ttl, line=row.line, ) provider_record_id = await provider.create_dns_record(dns_write) finally: await provider.close() return await self._finalize_create(row, provider_record_id) async def _finalize_create( self, row: CmdbDnsRecordRequest, provider_record_id: str, ) -> str | None: """Mirror an already-created cloud record into Helix. Split out of :meth:`_execute_create` so the executing-reaper can reuse it: when the reaper finds the cloud record present for a row stranded at EXECUTING, the write DID land and the only thing missing is this bookkeeping. Re-deriving it there would let the two paths drift on owner / recertification / owner-history provenance. """ from app.domains.cmdb.owner_history import ( CHANGE_REASON_DNS_REQUEST_CREATE, RESOURCE_TYPE_DNS_RECORD, record_owner_change, ) # Owner = requester (per the briefing). Recertification clock # starts now so the record has a full period before the first # nag. Pass both into the initial INSERT so the row is committed # atomically with full provenance — a separate post-commit # assignment would leave a window where the row exists with # NULL owner, and would orphan the row forever if the second # commit fails after the cloud-side record is already created. try: new_record = await CmdbService.create_dns_record_in_db( self.db, integration_id=row.integration_id, zone_name=row.zone_name, name=row.name, record_type=row.record_type, value=row.value, ttl=row.ttl, line=row.line, provider_record_id=provider_record_id, owner_person_id=str(row.requester_person_id), last_owner_review_at=datetime.now(UTC), actor_person_id=str(row.requester_person_id), ) except Exception: # Cloud record was already created — Helix DB insert failed. # Log the orphan loudly with everything an admin needs to # find and clean up the cloud row manually (provider can't # be auto-rolled back safely from inside an exception # handler — Tencent/Route53/DNSPod each have different # idempotency semantics on delete-by-record-id). logger.error( "DNS request %s: ORPHAN CLOUD RECORD — provider created " "record %s in zone %s on integration %s but Helix DB " "insert failed. Admin must delete the cloud record " "manually.", row.id, provider_record_id, row.zone_name, row.integration_id, ) raise # Emit owner-history seed for the new record. Previous = NULL, # new = requester, attributed to the requester (they are the # actor as well as the recipient). Use a fresh helper so # admin-side UI can render the "owned since DNS request X" link. await record_owner_change( self.db, resource_type=RESOURCE_TYPE_DNS_RECORD, resource_id=str(new_record.id), previous_owner_person_id=None, new_owner_person_id=str(row.requester_person_id), actor_person_id=str(row.requester_person_id), change_reason=CHANGE_REASON_DNS_REQUEST_CREATE, trace_id=row.feishu_instance_code, context={"dns_request_id": str(row.id)}, ) return str(new_record.id) async def _execute_update( self, row: CmdbDnsRecordRequest, ) -> str | None: # Round-30: lock the target row so a concurrent admin DELETE # (router.delete_dns_record) doesn't race the deleted_at check. # ``.get()`` was not lock-aware — the previous owner could # soft-delete the record between this check and the provider # write, leaving the cloud-side mutated and the DB row dead. target = (await self.db.execute( select(CmdbDnsRecord) .where(CmdbDnsRecord.id == _uuid.UUID(str(row.target_record_id))) .with_for_update() )).scalar_one_or_none() if target is None or target.deleted_at is not None: raise AppException( code="DNS_REQ_TARGET_GONE", message="Target record disappeared between approval and execution.", status_code=410, ) # Re-check ownership defensively. If the record was reassigned # away from the requester between submit and approval (e.g., # offboarding T+0 force-transfer fired), bail out. if str(target.owner_person_id or "") != str(row.requester_person_id): raise AppException( code="DNS_REQ_TARGET_OWNERSHIP_CHANGED", message=( "Target record ownership changed between submit and" " execution; refusing to apply." ), status_code=409, ) provider = await _load_dns_provider(self.db, row.integration_id) try: dns_write = DnsRecordWrite( zone_name=target.zone_name, name=row.name, record_type=row.record_type, value=row.value, ttl=row.ttl, line=row.line if row.line is not None else target.line, ) await provider.update_dns_record( target.record_id, target.zone_name, dns_write, ) finally: await provider.close() target.name = row.name target.value = row.value target.ttl = row.ttl if row.line is not None: target.line = row.line target.last_owner_review_at = datetime.now(UTC) # Approved-and-executed requests must stamp the modifier so the # UI's "修改人" column reflects the requester (not the admin # approver — the requester is the one who authored the change). CmdbService._stamp_modifier(target, str(row.requester_person_id)) await self.db.flush() await self.db.commit() return str(target.id) async def _execute_delete( self, row: CmdbDnsRecordRequest, ) -> str | None: # Round-30: lock the target row (see _execute_update). target = (await self.db.execute( select(CmdbDnsRecord) .where(CmdbDnsRecord.id == _uuid.UUID(str(row.target_record_id))) .with_for_update() )).scalar_one_or_none() if target is None or target.deleted_at is not None: # Target is already gone — treat as a no-op success so the # employee sees "done", but stamp ``execute_already_gone`` # in the audit detail so a forensic walk can distinguish # "I deleted it" from "someone else got there first". logger.info( "DNS request %s delete: target %s already gone", row.id, row.target_record_id, ) await self._audit( trace_id=f"feishu:{row.feishu_instance_code or row.id}", action="cmdb.dns_request.execute_target_already_gone", row=row, extra={"target_record_id": str(row.target_record_id)}, ) return None if str(target.owner_person_id or "") != str(row.requester_person_id): raise AppException( code="DNS_REQ_TARGET_OWNERSHIP_CHANGED", message=( "Target record ownership changed between submit and" " execution; refusing to apply." ), status_code=409, ) provider = await _load_dns_provider(self.db, row.integration_id) try: await provider.delete_dns_record( target.record_id, target.zone_name, ) finally: await provider.close() # Round-28: write an owner_history TERMINAL row so the admin # UI's ownership-timeline shows the record's lifecycle ending # explicitly. Without this the timeline reads "owned by X → # forever" even after deletion; auditors then can't tell from # the history alone whether the record was deleted or sync- # swept (which has its own terminal entry). from app.domains.cmdb.owner_history import ( CHANGE_REASON_DNS_REQUEST_DELETE, RESOURCE_TYPE_DNS_RECORD, record_owner_change, ) previous_owner = target.owner_person_id await CmdbService.delete_dns_record_in_db( self.db, str(target.id), actor_person_id=str(row.requester_person_id), ) await record_owner_change( self.db, resource_type=RESOURCE_TYPE_DNS_RECORD, resource_id=str(target.id), previous_owner_person_id=previous_owner, new_owner_person_id=None, actor_person_id=str(row.requester_person_id), change_reason=CHANGE_REASON_DNS_REQUEST_DELETE, trace_id=row.feishu_instance_code, context={ "dns_request_id": str(row.id), "action": "delete", }, ) return str(target.id) async def _execute_set_status( self, row: CmdbDnsRecordRequest, *, enable: bool, ) -> str | None: """Execute an approved ``enable`` or ``disable`` request. Mirrors the ``_execute_update`` defensive checks (target gone / ownership changed) — the cloud-side flip is non-destructive but we still don't want a requester whose access was revoked between submit and approval to slip an action through. Provider-level ``CapabilityNotSupported`` (AWS Route 53 has no enable/disable) bubbles up to the outer try/except and lands the row at ``failed`` with a clear error message. """ # Round-30: lock the target row (see _execute_update). target = (await self.db.execute( select(CmdbDnsRecord) .where(CmdbDnsRecord.id == _uuid.UUID(str(row.target_record_id))) .with_for_update() )).scalar_one_or_none() if target is None or target.deleted_at is not None: raise AppException( code="DNS_REQ_TARGET_GONE", message="Target record disappeared between approval and execution.", status_code=410, ) if str(target.owner_person_id or "") != str(row.requester_person_id): raise AppException( code="DNS_REQ_TARGET_OWNERSHIP_CHANGED", message=( "Target record ownership changed between submit and" " execution; refusing to apply." ), status_code=409, ) # Skip the cloud call if the record is already in the desired # state — saves a round-trip and avoids the DNSPod "no change" # warning that some accounts surface as an error. desired_status = "enable" if enable else "disable" if target.status == desired_status: logger.info( "DNS request %s set-status: target %s already in state %s — no-op", row.id, target.id, desired_status, ) return str(target.id) provider = await _load_dns_provider(self.db, row.integration_id) try: await provider.set_dns_record_status( target.record_id, target.zone_name, enable, ) finally: await provider.close() target.status = desired_status target.last_owner_review_at = datetime.now(UTC) # Stamp modifier so the UI shows who flipped the status, even # though the request flowed through Feishu approval (the admin # approver is implicit; the row's last_modified is the requester # who initiated the state change). CmdbService._stamp_modifier(target, str(row.requester_person_id)) await self.db.flush() await self.db.commit() return str(target.id) async def _audit( self, *, trace_id: str, action: str, row: CmdbDnsRecordRequest, extra: dict[str, Any] | None = None, ) -> None: audit = AuditService(self.db) detail: dict[str, Any] = { "request_id": str(row.id), "action": row.action, "zone_name": row.zone_name, "name": row.name, "record_type": row.record_type, "feishu_instance_code": row.feishu_instance_code, } if extra: detail.update(extra) await audit.log( trace_id=trace_id, actor_type="integration", actor_user_id=None, action=action, status="success", resource_type="cmdb_dns_record_request", resource_id=str(row.id), detail=detail, )