"""Auditor sync services — pull upstream state into auditor_* tables. One class per upstream concern; each ``sync()`` is idempotent and returns a typed summary dict so the scheduled-jobs runner and the ``/auditor/*/sync`` admin-trigger endpoints can render an honest "X records, Y new, Z stale" line. No fallbacks. If an integration is missing the sync raises ``IntegrationMissingError`` (caught at the router boundary and surfaced as a 503 with the auditor's ``audit_logs`` row carrying the real reason). Empty results are never silently produced from synthetic data — they reflect the actual upstream state. """ from __future__ import annotations import uuid from collections import defaultdict from dataclasses import dataclass from datetime import UTC, date, datetime, timedelta from typing import Any from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.geoip import GeoResult, haversine_km from app.core.geoip import lookup as geoip_lookup from app.core.safe_logging import get_safe_logger from app.domains.auditor.constants import ( ALL_MFA_PROVIDERS, ANOMALY_LEVEL_CRITICAL, ANOMALY_LEVEL_HIGH, ANOMALY_REASON_HEADLESS_UA, ANOMALY_REASON_IMPOSSIBLE_TRAVEL, ANOMALY_REASON_KNOWN_BAD_IP, ANOMALY_REASON_LOGIN_AFTER_OFFBOARD, ANOMALY_REASON_NEW_COUNTRY, ANOMALY_REASON_RAPID_FAILED_ATTEMPTS, ANOMALY_REASON_UNUSUAL_HOUR, CERT_KIND_DOMAIN, CERT_KIND_ICP, CERT_KIND_SSL, CERT_SOURCE_ALIYUN_DOMAIN, CERT_SOURCE_AWS_ACM, CERT_SOURCE_MANUAL, CERT_SOURCE_TENCENT_DOMAIN, CERT_STATUS_ACTIVE, CERT_STATUS_EXPIRED, DEFAULT_ROTATION_PERIOD_DAYS, EXPOSURE_SEVERITY_CRITICAL, EXPOSURE_SEVERITY_HIGH, EXPOSURE_SEVERITY_MEDIUM, EXPOSURE_SOURCE_ATLASSIAN, EXPOSURE_SOURCE_AWS_S3, EXPOSURE_SOURCE_AWS_SG, EXPOSURE_STATUS_OPEN, EXPOSURE_STATUS_RESOLVED, EXPOSURE_TYPE_ATLASSIAN_PUBLIC_PAGE, EXPOSURE_TYPE_INGRESS_ANY, EXPOSURE_TYPE_INGRESS_SENSITIVE, EXPOSURE_TYPE_S3_PUBLIC_ACL, EXPOSURE_TYPE_S3_PUBLIC_POLICY, MFA_PROVIDER_ATLASSIAN, MFA_PROVIDER_FEISHU, MFA_PROVIDER_GITHUB, MFA_PROVIDER_GOOGLE, PAT_BROAD_ADMIN_SCOPES, PAT_DORMANT_DAYS, PAT_KIND_ATLASSIAN_API_TOKEN, PAT_KIND_GITHUB_PAT, PAT_KIND_HELIX_API_KEY, PAT_KIND_HELIX_PAT, PAT_KIND_OAUTH_GRANT, PAT_RISK_BROAD_SCOPE_ADMIN, PAT_RISK_DORMANT, PAT_RISK_NO_EXPIRY, PAT_RISK_OVER_SCOPED, PAT_RISK_OWNER_LEFT, PAT_SOURCE_ATLASSIAN, PAT_SOURCE_GITHUB, PAT_SOURCE_HELIX_APIKEY, PAT_SOURCE_HELIX_OAUTH, PAT_SOURCE_HELIX_PAT, PAT_STATUS_ACTIVE, PAT_STATUS_DORMANT, PAT_STATUS_EXPIRED, PAT_STATUS_REVOKED, ROTATION_DUE_SOON_THRESHOLD_DAYS, SECRET_KIND_API_KEY, SECRET_KIND_CLOUD_ACCESS_KEY, SECRET_KIND_OAUTH_CLIENT_SECRET, SECRET_KIND_SSL_CERT, SECRET_SOURCE_ALIYUN, SECRET_SOURCE_AWS, SECRET_SOURCE_AWS_ACM, SECRET_SOURCE_AZURE, SECRET_SOURCE_HELIX_APIKEY, SECRET_SOURCE_TENCENT, SECRET_STATUS_CURRENT, SECRET_STATUS_DUE_SOON, SECRET_STATUS_OVERDUE, SENSITIVE_PORTS, SIGNIN_PROVIDER_FEISHU, SIGNIN_PROVIDER_GOOGLE, SIGNIN_STATUS_FAILURE, anomaly_level_for_score, cert_alert_level_for_days, ) from app.domains.auditor.integration_adapters import ( AliyunAuditorAdapter, AtlassianAuditorAdapter, AwsAuditorAdapter, AzureAuditorAdapter, FeishuAuditorAdapter, GitHubAuditorAdapter, GoogleAuditorAdapter, IntegrationMissingError, TencentAuditorAdapter, ) from app.domains.auditor.models import ( AuditorCertExpiry, AuditorMfaStatus, AuditorPatAudit, AuditorPublicExposureFinding, AuditorSecretRotationRecord, AuditorSigninEvent, ) from app.identity.person_models import IdentityPerson logger = get_safe_logger(__name__) # ───────────────────────────────────────────────────────────────────── # Helpers # ───────────────────────────────────────────────────────────────────── def _utc_now() -> datetime: return datetime.now(UTC) def _aware(dt: datetime | None) -> datetime | None: """Coerce a tz-naive datetime to UTC-aware. asyncmy + MySQL ``TIMESTAMP`` columns return tz-naive Python datetimes even when the SQLAlchemy column is declared ``DateTime(timezone=True)``. Subtracting them from a tz-aware ``synced_at`` raises ``TypeError: can't subtract offset-naive and offset-aware datetimes`` — exactly the error that surfaces in ``last_error`` for ``secret_rotation`` and ``pat_audit`` cron runs. Every DB-sourced datetime that participates in arithmetic with ``synced_at`` must pass through this helper. Returns ``None`` unchanged so callers don't have to repeat the null check. """ if dt is None: return None if dt.tzinfo is None: return dt.replace(tzinfo=UTC) return dt def _json_safe(obj: Any) -> Any: """Recursively convert upstream API responses into JSON-serializable values. boto3 / aliyun SDK / google-api-python-client / etc. routinely embed ``datetime`` and ``date`` objects in their response dicts (e.g. AWS IAM ``CreateDate`` / ``LastUsedDate``, ACM ``NotBefore`` / ``NotAfter``, S3 ``CreationDate``). Stuffing those into a JSON column raises ``TypeError: Object of type datetime is not JSON serializable`` at flush time and poisons the whole transaction. Applied at every upsert helper that writes a JSON column so the adapters don't have to remember. """ if isinstance(obj, datetime): return obj.isoformat() if isinstance(obj, date): return obj.isoformat() if isinstance(obj, dict): return {k: _json_safe(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [_json_safe(v) for v in obj] if isinstance(obj, (bytes, bytearray)): return obj.decode("utf-8", errors="replace") return obj async def _persons_by_email(session: AsyncSession) -> dict[str, IdentityPerson]: """Return ``{lower_email: person}`` over every live person. Two invariants this map has to hold, both of which it used to break: 1. **Tombstones are excluded.** Soft-deleted rows in this codebase keep ``employment_status='active'`` (purged mis-created hires, stale-ingest duplicates), so ``deleted_at IS NULL`` is the only filter that works. A tombstone winning an email meant the MFA / sign-in / PAT rows attached to a person_id that every dashboard query filters out — the employee's MFA state simply vanished. 2. **Collisions resolve deterministically.** When a tombstone and a live row share ``work_email``, the live row must win regardless of scan order — hence the explicit ORDER BY rather than last-writer-wins over an unordered result set. """ rows = (await session.execute( select(IdentityPerson) .where(IdentityPerson.deleted_at.is_(None)) # Oldest first, so the newest live row is the last writer and # wins the slot; stable across runs. .order_by(IdentityPerson.created_at.asc(), IdentityPerson.id.asc()) )).scalars().all() out: dict[str, IdentityPerson] = {} for p in rows: if p.work_email: out[p.work_email.strip().lower()] = p if p.personal_email: out.setdefault(p.personal_email.strip().lower(), p) return out def _by_local_part( emails: dict[str, IdentityPerson], ) -> dict[str, IdentityPerson]: """``{email_local_part: person}`` — built once per sync (#2188-2). The IAM-key and GitHub-PAT owner matchers each walked the WHOLE person→email map for EVERY token, comparing local parts: O(persons × tokens) string work per run, for a lookup that is a dict hit. First writer wins so the precedence matches ``_persons_by_email``'s own ordering (live rows, oldest first). """ out: dict[str, IdentityPerson] = {} for email, person in emails.items(): local, _, _ = email.partition("@") if local: out.setdefault(local, person) return out def _record_provider_error( errors: list[dict[str, Any]], *, provider_label: str | None = None, source_label: str | None = None, exc: Exception, ) -> None: """Append a normalized per-upstream error entry. Two cases that must both be survivable for the auditor's contract that one provider's failure never 500s the whole sync: * ``IntegrationMissingError`` — integration not configured / scope absent. Expected at runtime, recorded at INFO level. * Any other ``Exception`` — upstream API outage, JSON shape drift, timeout, etc. Recorded at WARNING level with traceback so ops can see what actually broke without the user hitting a 500. The shape preserves the legacy ``error`` field used by the frontend, plus a ``kind`` discriminator and the original exception type for triage. """ is_missing = isinstance(exc, IntegrationMissingError) if is_missing: logger.info( "auditor sync skipped (integration missing) %s: %s", provider_label or source_label, exc, ) else: logger.warning( "auditor sync upstream error %s: %s", provider_label or source_label, exc, exc_info=True, ) entry: dict[str, Any] = { "error": ( str(exc) if is_missing else f"{type(exc).__name__}: {exc}" ), "kind": "integration_missing" if is_missing else "upstream_error", } if provider_label is not None: entry["provider"] = provider_label if source_label is not None: entry["source"] = source_label errors.append(entry) # ═════════════════════════════════════════════════════════════════════ # 1. MFA sync # ═════════════════════════════════════════════════════════════════════ class MfaSyncService: """Pull MFA enrollment status from each provider and upsert into ``auditor_mfa_status``. Per-provider semantics: * Google: ``users.list(projection=full).isEnrolledIn2Sv`` * Feishu: ``contact.v3.user`` ``security_strategy.two_factor_enabled`` (NULL when scope absent — recorded as ``is_enrolled=False`` with a flag in raw_payload) * Atlassian: not in this PR — Helix has no ``atlassian_admin`` integration kind. The MFA dashboard row for atlassian comes from manual-entry rows the auditor populates (out of scope for the SYNC path). * GitHub: ``GET /orgs/{org}/members?role=admin`` plus ``has_two_factor_authentication`` if available. """ @staticmethod async def sync( session: AsyncSession, *, providers: list[str] | None = None, actor_user_id: uuid.UUID | None = None, trace_id: uuid.UUID | None = None, ) -> dict[str, Any]: providers = providers or list(ALL_MFA_PROVIDERS) emails = await _persons_by_email(session) synced_at = _utc_now() per_provider: dict[str, dict[str, int]] = {} errors: list[dict[str, Any]] = [] if MFA_PROVIDER_GOOGLE in providers: try: rows = await GoogleAuditorAdapter.list_users_mfa_status( session, ) except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error(errors, provider_label="google", exc=exc) rows = [] counts = {"total": 0, "enrolled": 0, "matched_persons": 0} for g_row in rows: counts["total"] += 1 if g_row.is_enrolled_in_2sv: counts["enrolled"] += 1 person = emails.get(g_row.user_email.lower()) if person is None: continue counts["matched_persons"] += 1 await _upsert_mfa( session, person_id=person.id, provider=MFA_PROVIDER_GOOGLE, provider_user_id=g_row.google_user_id, provider_email=g_row.user_email, is_enrolled=g_row.is_enrolled_in_2sv, is_enforced=g_row.is_enforced_in_2sv, # Directory API answers both flags for every user. enrollment_observed=True, enforcement_observed=True, enrollment_methods=None, # Google's Directory API does NOT expose a # "last 2-step verification" timestamp. The user's # last_login_at conflates browser-cookie sessions # with explicit MFA prompts, so we leave this NULL # rather than misleading auditors. last_verified_at=None, last_synced_at=synced_at, raw_payload={"google": g_row.raw}, ) per_provider["google"] = counts if MFA_PROVIDER_FEISHU in providers: try: rows_f = await FeishuAuditorAdapter.list_users_mfa_status( session, ) except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error(errors, provider_label="feishu", exc=exc) rows_f = [] counts = {"total": 0, "enrolled": 0, "matched_persons": 0} for f_row in rows_f: counts["total"] += 1 if f_row.is_2step_enabled: counts["enrolled"] += 1 if not f_row.user_email: continue person = emails.get((f_row.user_email or "").lower()) if person is None: continue counts["matched_persons"] += 1 await _upsert_mfa( session, person_id=person.id, provider=MFA_PROVIDER_FEISHU, provider_user_id=f_row.open_id, provider_email=f_row.user_email, is_enrolled=f_row.is_2step_enabled, is_enforced=False, # ``security_strategy`` is absent for tenants without # enterprise + audit scope — then ``is_2step_enabled`` # is a default, not a reading. enrollment_observed=f_row.is_2step_observed, # Feishu exposes no tenant-level MFA enforcement flag. enforcement_observed=False, enrollment_methods=None, last_verified_at=None, last_synced_at=synced_at, raw_payload={"feishu": f_row.raw}, ) per_provider["feishu"] = counts if MFA_PROVIDER_ATLASSIAN in providers: try: rows_at = await AtlassianAuditorAdapter.list_users_mfa_status( session, ) except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, provider_label="atlassian", exc=exc, ) rows_at = [] counts = {"total": 0, "enrolled": 0, "matched_persons": 0} for at_row in rows_at: counts["total"] += 1 if at_row.is_enrolled: counts["enrolled"] += 1 if not at_row.user_email: continue person = emails.get(at_row.user_email.lower()) if person is None: continue counts["matched_persons"] += 1 await _upsert_mfa( session, person_id=person.id, provider=MFA_PROVIDER_ATLASSIAN, provider_user_id=at_row.user_account_id, provider_email=at_row.user_email, is_enrolled=at_row.is_enrolled, # ``policy_status='enforced'`` means the org enforces # 2-step verification on this user's directory; that # IS enforcement. Any other value means non-enforced. is_enforced=(at_row.policy_status == "enforced"), # Atlassian has no per-user 2FA attribute at all: both # facts are derived from the org authentication policy, # so when that policy cannot be read # (``policy_absent``) NEITHER fact was observed. enrollment_observed=(at_row.policy_status != "policy_absent"), enforcement_observed=(at_row.policy_status != "policy_absent"), enrollment_methods=None, last_verified_at=None, last_synced_at=synced_at, raw_payload={ "atlassian": at_row.raw, "policy_status": at_row.policy_status, "policy_directory_id": at_row.policy_directory_id, }, ) per_provider["atlassian"] = counts if MFA_PROVIDER_GITHUB in providers: try: gh_rows = await GitHubAuditorAdapter.list_organization_members_with_mfa( session, ) except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, provider_label="github", exc=exc, ) gh_rows = [] counts = { "total": len(gh_rows), "enrolled": 0, "matched_persons": 0, } # GitHub members are matched to Helix persons by the # local-part of work_email == GitHub login (the standard # PingCAP convention: ``john@pingcap.com`` ↔ login ``john``). # Members whose login does NOT map to a Helix person are # recorded only if the integration carries an explicit # ``login_to_email`` mapping; otherwise skipped silently — # auditor_mfa_status.person_id is NOT NULL by design. login_index: dict[str, IdentityPerson] = {} for email, p in emails.items(): local, _, _ = email.partition("@") if local: login_index.setdefault(local.lower(), p) for gh_row in gh_rows: if gh_row.is_2fa_enabled: counts["enrolled"] += 1 person = login_index.get(gh_row.user_login.lower()) if person is None: continue counts["matched_persons"] += 1 await _upsert_mfa( session, person_id=person.id, provider=MFA_PROVIDER_GITHUB, provider_user_id=gh_row.user_login, provider_email=gh_row.user_email, is_enrolled=gh_row.is_2fa_enabled, # GitHub org SAML SSO + 2FA enforcement is a separate # toggle; we cannot read it from this endpoint. is_enforced=False, enrollment_observed=True, enforcement_observed=False, enrollment_methods=None, last_verified_at=None, last_synced_at=synced_at, raw_payload={ "github": gh_row.raw, "is_admin": gh_row.role == "admin", }, ) per_provider["github"] = counts await session.flush() from app.domains.auditor import findings_service as _findings await _findings.reconcile_mfa_findings(session) await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="auditor.mfa.sync", resource_type="auditor_mfa_status", status="success", detail={ "providers": providers, "per_provider": per_provider, "errors": errors, }, ) return { "providers": providers, "per_provider": per_provider, "errors": errors, "person_count": len(emails), "synced_at": synced_at, } async def _upsert_mfa( session: AsyncSession, *, person_id: uuid.UUID, provider: str, provider_user_id: str | None, provider_email: str | None, is_enrolled: bool, is_enforced: bool, # Whether the provider actually ANSWERED each question — see the # column comments on ``AuditorMfaStatus``. Required (no default) so a # future provider branch cannot silently inherit "observed". enrollment_observed: bool, enforcement_observed: bool, enrollment_methods: list[str] | None, last_verified_at: datetime | None, last_synced_at: datetime, raw_payload: dict[str, Any] | None, ) -> None: raw_payload = _json_safe(raw_payload) existing = ( await session.execute( select(AuditorMfaStatus).where( AuditorMfaStatus.person_id == person_id, AuditorMfaStatus.provider == provider, ) ) ).scalar_one_or_none() if existing is None: session.add(AuditorMfaStatus( id=uuid.uuid4(), person_id=person_id, provider=provider, provider_user_id=provider_user_id, provider_email=provider_email, is_enrolled=is_enrolled, is_enforced=is_enforced, enrollment_observed=enrollment_observed, enforcement_observed=enforcement_observed, enrollment_methods=enrollment_methods, last_verified_at=last_verified_at, last_synced_at=last_synced_at, raw_payload=raw_payload, )) return existing.provider_user_id = provider_user_id or existing.provider_user_id existing.provider_email = provider_email or existing.provider_email existing.is_enrolled = is_enrolled existing.is_enforced = is_enforced # Plain assignment, never ``or existing.…``: observability can go # from True back to False (a scope revoked, a policy deleted) and a # stale True would keep a blind row inside the coverage denominator. existing.enrollment_observed = enrollment_observed existing.enforcement_observed = enforcement_observed existing.enrollment_methods = enrollment_methods existing.last_verified_at = last_verified_at or existing.last_verified_at existing.last_synced_at = last_synced_at existing.raw_payload = raw_payload or existing.raw_payload # ═════════════════════════════════════════════════════════════════════ # 2. Public exposure scan # ═════════════════════════════════════════════════════════════════════ class PublicExposureScanService: """Scan AWS Security Groups + S3 buckets for unintended public exposure; upsert findings. """ @staticmethod async def scan( session: AsyncSession, *, actor_user_id: uuid.UUID | None = None, trace_id: uuid.UUID | None = None, ) -> dict[str, Any]: scan_run_id = uuid.uuid4() started_at = _utc_now() sources_scanned: list[str] = [] new_findings = 0 updated_findings = 0 resolved_findings = 0 errors: list[dict[str, Any]] = [] # A source only enters ``sources_scanned`` when the round covered # it COMPLETELY — that list is the sole gate on auto-resolve, so a # partially-scanned source (one AWS account with rotated keys, a # bucket whose ACL and policy reads both 403'd) must stay out of # it or every finding it owns gets closed as "not seen". partial_sources: dict[str, list[str]] = {} def _mark_scanned(source: str, partial: list[str]) -> None: if partial: partial_sources[source] = list(partial) _record_provider_error( errors, source_label=source, exc=RuntimeError( "partial scan — auto-resolve skipped: " + "; ".join(partial[:10]) ), ) return sources_scanned.append(source) # ─ AWS Security Groups ───────────────────────────────────── sg_partial: list[str] = [] try: sg_findings = await AwsAuditorAdapter.list_security_groups_with_open_ingress( session, partial=sg_partial, ) _mark_scanned(EXPOSURE_SOURCE_AWS_SG, sg_partial) except Exception as exc: # noqa: BLE001 — provider isolation sg_findings = [] _record_provider_error( errors, source_label=EXPOSURE_SOURCE_AWS_SG, exc=exc, ) seen_sg_ids: set[tuple[str, str, str]] = set() for f in sg_findings: for rule in f.ingress_rules_open: from_port = rule.get("from_port") to_port = rule.get("to_port") # Interval intersection, not materialisation (#2188-4). # A rule opening 0-65535 (which is exactly the rule most # worth flagging) used to build a 65 536-element list per # rule per scan just to ask "does this range contain any # sensitive port?". touches_sensitive = (from_port is None and to_port is None) if not touches_sensitive and ( from_port is not None and to_port is not None ): lo, hi = int(from_port), int(to_port) if lo > hi: lo, hi = hi, lo touches_sensitive = any( lo <= p <= hi for p in SENSITIVE_PORTS ) exposure_type = ( EXPOSURE_TYPE_INGRESS_SENSITIVE if touches_sensitive else EXPOSURE_TYPE_INGRESS_ANY ) severity = ( EXPOSURE_SEVERITY_CRITICAL if touches_sensitive else EXPOSURE_SEVERITY_MEDIUM ) resource_key = ( EXPOSURE_SOURCE_AWS_SG, f.sg_id, exposure_type, ) seen_sg_ids.add(resource_key) created, updated = await _upsert_exposure( session, scan_run_id=scan_run_id, source=EXPOSURE_SOURCE_AWS_SG, region=f.region, resource_id=f.sg_id, resource_arn=None, resource_name=f.sg_name, exposure_type=exposure_type, exposure_detail={"rule": rule, "vpc_id": f.vpc_id}, severity=severity, raw_payload={"sg": f.raw, "rule": rule}, seen_at=started_at, ) new_findings += int(created) updated_findings += int(updated and not created) # ─ AWS S3 public buckets ─────────────────────────────────── s3_partial: list[str] = [] try: s3_findings = await AwsAuditorAdapter.list_s3_public_buckets( session, partial=s3_partial, ) _mark_scanned(EXPOSURE_SOURCE_AWS_S3, s3_partial) except Exception as exc: # noqa: BLE001 — provider isolation s3_findings = [] _record_provider_error( errors, source_label=EXPOSURE_SOURCE_AWS_S3, exc=exc, ) seen_s3_keys: set[tuple[str, str, str]] = set() for f in s3_findings: if f.public_acl: key = (EXPOSURE_SOURCE_AWS_S3, f.bucket_name, EXPOSURE_TYPE_S3_PUBLIC_ACL) seen_s3_keys.add(key) created, updated = await _upsert_exposure( session, scan_run_id=scan_run_id, source=EXPOSURE_SOURCE_AWS_S3, region=f.region, resource_id=f.bucket_name, resource_arn=f"arn:aws:s3:::{f.bucket_name}", resource_name=f.bucket_name, exposure_type=EXPOSURE_TYPE_S3_PUBLIC_ACL, exposure_detail={"public_acl": True}, severity=EXPOSURE_SEVERITY_HIGH, raw_payload={"bucket": f.raw}, seen_at=started_at, ) new_findings += int(created) updated_findings += int(updated and not created) if f.public_policy: key = (EXPOSURE_SOURCE_AWS_S3, f.bucket_name, EXPOSURE_TYPE_S3_PUBLIC_POLICY) seen_s3_keys.add(key) created, updated = await _upsert_exposure( session, scan_run_id=scan_run_id, source=EXPOSURE_SOURCE_AWS_S3, region=f.region, resource_id=f.bucket_name, resource_arn=f"arn:aws:s3:::{f.bucket_name}", resource_name=f.bucket_name, exposure_type=EXPOSURE_TYPE_S3_PUBLIC_POLICY, exposure_detail={"public_policy": True}, severity=EXPOSURE_SEVERITY_HIGH, raw_payload={"bucket": f.raw}, seen_at=started_at, ) new_findings += int(created) updated_findings += int(updated and not created) # ─ Atlassian public spaces / projects ────────────────────── try: at_findings = await AtlassianAuditorAdapter.list_public_spaces( session, ) sources_scanned.append(EXPOSURE_SOURCE_ATLASSIAN) except Exception as exc: # noqa: BLE001 — provider isolation at_findings = [] _record_provider_error( errors, source_label=EXPOSURE_SOURCE_ATLASSIAN, exc=exc, ) seen_at_keys: set[tuple[str, str, str]] = set() for f in at_findings: key = ( EXPOSURE_SOURCE_ATLASSIAN, f"{f.space_type}:{f.space_key}", EXPOSURE_TYPE_ATLASSIAN_PUBLIC_PAGE, ) seen_at_keys.add(key) created, updated = await _upsert_exposure( session, scan_run_id=scan_run_id, source=EXPOSURE_SOURCE_ATLASSIAN, region=None, resource_id=f"{f.space_type}:{f.space_key}", resource_arn=None, resource_name=f.space_name or f.space_key, exposure_type=EXPOSURE_TYPE_ATLASSIAN_PUBLIC_PAGE, exposure_detail={ "space_type": f.space_type, "site_url": f.site_url, "public_url": f.public_url, }, # Atlassian anonymous read = high (it's by design that # the page is world-readable, but the auditor surfaces # it so a human reviews whether that's intentional). severity=EXPOSURE_SEVERITY_HIGH, raw_payload=f.raw, seen_at=started_at, ) new_findings += int(created) updated_findings += int(updated and not created) # ─ Auto-resolve findings no longer seen ──────────────────── # ONLY for sources we scanned COMPLETELY. If the scan failed — # or covered only part of the fleet — we leave findings open. async def _resolve_if_complete( source: str, seen_keys: set[tuple[str, str, str]], ) -> int: if source not in sources_scanned: return 0 if partial_sources.get(source): # Defence in depth: _mark_scanned already refuses to add # a partial source, so reaching here means someone added # a new append path that bypassed the gate. logger.error( "auto-resolve refused for %s — partial scan: %s", source, partial_sources[source], ) return 0 return await _auto_resolve_unseen( session, source=source, seen_keys=seen_keys, resolved_at=started_at, resolution_note="not_seen_in_scan", ) resolved_findings += await _resolve_if_complete( EXPOSURE_SOURCE_AWS_SG, seen_sg_ids, ) resolved_findings += await _resolve_if_complete( EXPOSURE_SOURCE_AWS_S3, seen_s3_keys, ) resolved_findings += await _resolve_if_complete( EXPOSURE_SOURCE_ATLASSIAN, seen_at_keys, ) await session.flush() # Project the refreshed exposure rows into the unified worklist. from app.domains.auditor import findings_service as _findings await _findings.project_exposure_findings(session) completed_at = _utc_now() total_findings = new_findings + updated_findings await AuditService(session).log( trace_id=trace_id or scan_run_id, actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="auditor.public_exposure.scan", resource_type="auditor_public_exposure_findings", status="success", detail={ "scan_run_id": str(scan_run_id), "sources_scanned": sources_scanned, "partial_sources": partial_sources, "findings_count": total_findings, "new": new_findings, "updated": updated_findings, "resolved": resolved_findings, "errors": errors, }, ) return { "scan_run_id": scan_run_id, "sources_scanned": sources_scanned, "partial_sources": partial_sources, "findings_count": total_findings, "started_at": started_at, "completed_at": completed_at, "errors": errors, } async def _upsert_exposure( session: AsyncSession, *, scan_run_id: uuid.UUID, source: str, region: str | None, resource_id: str, resource_arn: str | None, resource_name: str | None, exposure_type: str, exposure_detail: dict[str, Any], severity: str, raw_payload: dict[str, Any], seen_at: datetime, ) -> tuple[bool, bool]: """Returns ``(created, updated)``.""" exposure_detail = _json_safe(exposure_detail) raw_payload = _json_safe(raw_payload) existing = ( await session.execute( select(AuditorPublicExposureFinding).where( AuditorPublicExposureFinding.source == source, AuditorPublicExposureFinding.resource_id == resource_id, AuditorPublicExposureFinding.exposure_type == exposure_type, ) ) ).scalar_one_or_none() if existing is None: session.add(AuditorPublicExposureFinding( id=uuid.uuid4(), scan_run_id=scan_run_id, source=source, region=region, resource_id=resource_id, resource_arn=resource_arn, resource_name=resource_name, exposure_type=exposure_type, exposure_detail=exposure_detail, severity=severity, first_seen_at=seen_at, last_seen_at=seen_at, status=EXPOSURE_STATUS_OPEN, raw_payload=raw_payload, )) return True, False # Refresh the finding — bumps last_seen_at & severity if escalated existing.last_seen_at = seen_at existing.exposure_detail = exposure_detail existing.severity = severity existing.scan_run_id = scan_run_id existing.region = region or existing.region existing.resource_arn = resource_arn or existing.resource_arn existing.resource_name = resource_name or existing.resource_name existing.raw_payload = raw_payload if existing.status == EXPOSURE_STATUS_RESOLVED: # Re-opened: same exposure detected again existing.status = EXPOSURE_STATUS_OPEN existing.resolved_at = None existing.resolved_by_person_id = None existing.resolution_note = None return False, True async def _auto_resolve_unseen( session: AsyncSession, *, source: str, seen_keys: set[tuple[str, str, str]], resolved_at: datetime, resolution_note: str, ) -> int: """Mark every open finding from ``source`` not in ``seen_keys`` as ``status='resolved'``. Counts the number flipped. """ open_rows = (await session.execute( select(AuditorPublicExposureFinding).where( AuditorPublicExposureFinding.source == source, AuditorPublicExposureFinding.status == EXPOSURE_STATUS_OPEN, ) )).scalars().all() n = 0 for r in open_rows: key = (r.source, r.resource_id, r.exposure_type) if key in seen_keys: continue r.status = EXPOSURE_STATUS_RESOLVED r.resolved_at = resolved_at r.resolution_note = resolution_note n += 1 return n # ═════════════════════════════════════════════════════════════════════ # 3. Secret rotation ledger # ═════════════════════════════════════════════════════════════════════ class SecretRotationSyncService: """Refresh ``auditor_secret_rotation_records`` from: * Helix's own ``apikey_keys`` (per-user API keys) * AWS IAM access keys (per-user keys in every configured AWS account) * AWS ACM certificates (treated as a kind of secret with a renewal cadence — ALSO populates ``auditor_cert_expiries`` via the cert sync; here they're tracked from the rotation perspective) """ @staticmethod async def sync( session: AsyncSession, *, actor_user_id: uuid.UUID | None = None, trace_id: uuid.UUID | None = None, ) -> dict[str, Any]: synced_at = _utc_now() scanned_sources: list[str] = [] record_count = 0 errors: list[dict[str, Any]] = [] # ─ Helix API key assignments ────────────────────────────── try: from app.domains.apikey.models import ApiKeyAssignment apikey_rows = (await session.execute( select(ApiKeyAssignment).where( ApiKeyAssignment.status.notin_( ("revoked", "rejected", "cancelled") ), ) )).scalars().all() scanned_sources.append(SECRET_SOURCE_HELIX_APIKEY) for k in apikey_rows: period_days = DEFAULT_ROTATION_PERIOD_DAYS[ SECRET_KIND_API_KEY ] # Last rotated is the assignment created_at — keys are # immutable once issued; rotation = re-issue. # MySQL TIMESTAMP returns naive — coerce to aware so the # arithmetic below doesn't raise. last_rotated = _aware(k.created_at) # Use ``valid_until`` if set; otherwise default 90d # period from issue. if k.valid_until: next_due = datetime( k.valid_until.year, k.valid_until.month, k.valid_until.day, tzinfo=UTC, ) elif last_rotated: next_due = last_rotated + timedelta(days=period_days) else: next_due = None days_until = ( int((next_due - synced_at).total_seconds() // 86400) if next_due else None ) status = _status_for_days(days_until) await _upsert_secret( session, secret_kind=SECRET_KIND_API_KEY, source=SECRET_SOURCE_HELIX_APIKEY, resource_id=str(k.id), resource_name=k.project_name, owner_person_id=k.assigned_to, owner_email=None, created_at_source=k.created_at, last_rotated_at=last_rotated, rotation_period_days=period_days, next_rotation_due_at=next_due, status=status, days_until_due=days_until, raw_payload={ "request_number": k.request_number, "id": str(k.id), "project_name": k.project_name, }, last_synced_at=synced_at, ) record_count += 1 except ImportError: errors.append({ "source": SECRET_SOURCE_HELIX_APIKEY, "error": "apikey domain not installed", }) # ─ AWS IAM access keys ──────────────────────────────────── try: iam_keys = await AwsAuditorAdapter.list_iam_access_keys(session) scanned_sources.append(SECRET_SOURCE_AWS) emails = await _persons_by_email(session) emails_by_local = _by_local_part(emails) for k in iam_keys: period_days = DEFAULT_ROTATION_PERIOD_DAYS[ SECRET_KIND_CLOUD_ACCESS_KEY ] last_rotated = k.create_date next_due = ( last_rotated + timedelta(days=period_days) if last_rotated else None ) days_until = ( int((next_due - synced_at).total_seconds() // 86400) if next_due else None ) status = _status_for_days(days_until) # Best-effort owner email match by IAM user_name # (which is often the email-prefix or full email). username_lower = (k.user_name or "").lower() owner_person = ( emails.get(username_lower) or emails_by_local.get(username_lower) ) await _upsert_secret( session, secret_kind=SECRET_KIND_CLOUD_ACCESS_KEY, source=SECRET_SOURCE_AWS, resource_id=k.access_key_id, resource_name=f"iam/{k.user_name}", owner_person_id=owner_person.id if owner_person else None, owner_email=( owner_person.work_email if owner_person else None ), created_at_source=k.create_date, last_rotated_at=last_rotated, rotation_period_days=period_days, next_rotation_due_at=next_due, status=status, days_until_due=days_until, raw_payload=k.raw, last_synced_at=synced_at, ) record_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=SECRET_SOURCE_AWS, exc=exc, ) # ─ AWS ACM (rotation perspective) ───────────────────────── try: acm = await AwsAuditorAdapter.list_acm_certs(session) scanned_sources.append(SECRET_SOURCE_AWS_ACM) period_days = DEFAULT_ROTATION_PERIOD_DAYS[SECRET_KIND_SSL_CERT] for c in acm: last_rotated = c.not_before next_due = c.not_after days_until = ( int((next_due - synced_at).total_seconds() // 86400) if next_due else None ) status = _status_for_days(days_until) await _upsert_secret( session, secret_kind=SECRET_KIND_SSL_CERT, source=SECRET_SOURCE_AWS_ACM, resource_id=c.arn, resource_name=c.domain_name, owner_person_id=None, owner_email=None, created_at_source=c.not_before, last_rotated_at=last_rotated, rotation_period_days=period_days, next_rotation_due_at=next_due, status=status, days_until_due=days_until, raw_payload=c.raw, last_synced_at=synced_at, ) record_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=SECRET_SOURCE_AWS_ACM, exc=exc, ) # ─ Aliyun RAM access keys ───────────────────────────────── try: aly_keys = await AliyunAuditorAdapter.list_ram_access_keys(session) scanned_sources.append(SECRET_SOURCE_ALIYUN) period_days = DEFAULT_ROTATION_PERIOD_DAYS[ SECRET_KIND_CLOUD_ACCESS_KEY ] for k in aly_keys: last_rotated = k.create_date next_due = ( last_rotated + timedelta(days=period_days) if last_rotated else None ) days_until = ( int((next_due - synced_at).total_seconds() // 86400) if next_due else None ) status = _status_for_days(days_until) await _upsert_secret( session, secret_kind=SECRET_KIND_CLOUD_ACCESS_KEY, source=SECRET_SOURCE_ALIYUN, resource_id=k.access_key_id, resource_name=f"ram/{k.user_name}", owner_person_id=None, owner_email=None, created_at_source=k.create_date, last_rotated_at=last_rotated, rotation_period_days=period_days, next_rotation_due_at=next_due, status=status, days_until_due=days_until, raw_payload=k.raw, last_synced_at=synced_at, ) record_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=SECRET_SOURCE_ALIYUN, exc=exc, ) # ─ Tencent CAM access keys ──────────────────────────────── try: tc_keys = await TencentAuditorAdapter.list_cam_access_keys(session) scanned_sources.append(SECRET_SOURCE_TENCENT) period_days = DEFAULT_ROTATION_PERIOD_DAYS[ SECRET_KIND_CLOUD_ACCESS_KEY ] for k in tc_keys: last_rotated = k.create_time next_due = ( last_rotated + timedelta(days=period_days) if last_rotated else None ) days_until = ( int((next_due - synced_at).total_seconds() // 86400) if next_due else None ) status = _status_for_days(days_until) await _upsert_secret( session, secret_kind=SECRET_KIND_CLOUD_ACCESS_KEY, source=SECRET_SOURCE_TENCENT, resource_id=k.secret_id, resource_name=f"cam/{k.user_name}", owner_person_id=None, owner_email=None, created_at_source=k.create_time, last_rotated_at=last_rotated, rotation_period_days=period_days, next_rotation_due_at=next_due, status=status, days_until_due=days_until, raw_payload=k.raw, last_synced_at=synced_at, ) record_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=SECRET_SOURCE_TENCENT, exc=exc, ) # ─ Azure storage-account keys + SP secrets ──────────────── # Two distinct kinds: # * storage-account key: rotated server-side via listKeys; we # read ``creationTime`` to know when it was last rotated. # * SP password credential (client_secret): has explicit # end_date; we treat ``end_date`` as ``next_rotation_due_at`` # so an SP secret that's about to expire shows as due_soon # even when the rotation_period heuristic isn't appropriate. try: az_storage = await AzureAuditorAdapter.list_storage_account_keys( session, ) scanned_sources.append(SECRET_SOURCE_AZURE) period_days = DEFAULT_ROTATION_PERIOD_DAYS[ SECRET_KIND_CLOUD_ACCESS_KEY ] for k in az_storage: last_rotated = k.last_rotated_at next_due = ( last_rotated + timedelta(days=period_days) if last_rotated else None ) days_until = ( int((next_due - synced_at).total_seconds() // 86400) if next_due else None ) status = _status_for_days(days_until) resource_id = ( f"{k.subscription_id}/{k.resource_group}/" f"{k.storage_account_name}/{k.key_name}" ) await _upsert_secret( session, secret_kind=SECRET_KIND_CLOUD_ACCESS_KEY, source=SECRET_SOURCE_AZURE, resource_id=resource_id, resource_name=( f"storage/{k.storage_account_name}/{k.key_name}" ), owner_person_id=None, owner_email=None, created_at_source=k.last_rotated_at, last_rotated_at=last_rotated, rotation_period_days=period_days, next_rotation_due_at=next_due, status=status, days_until_due=days_until, raw_payload=k.raw, last_synced_at=synced_at, ) record_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=SECRET_SOURCE_AZURE, exc=exc, ) try: az_sp = await AzureAuditorAdapter.list_app_password_credentials( session, ) # SP secrets are part of the Azure source bucket; only mark # ``scanned_sources`` once even if both calls succeed. if SECRET_SOURCE_AZURE not in scanned_sources: scanned_sources.append(SECRET_SOURCE_AZURE) sp_period = DEFAULT_ROTATION_PERIOD_DAYS[ SECRET_KIND_OAUTH_CLIENT_SECRET ] for c in az_sp: last_rotated = c.start_date # Use the explicit end_date from the SP credential; it's # the *real* expiry. Fallback to start_date + period only # when end_date is absent (rare — Azure forces an end # date at create time). next_due = c.end_date or ( (last_rotated + timedelta(days=sp_period)) if last_rotated else None ) days_until = ( int((next_due - synced_at).total_seconds() // 86400) if next_due else None ) status = _status_for_days(days_until) resource_id = f"{c.application_id}/{c.credential_id}" await _upsert_secret( session, secret_kind=SECRET_KIND_OAUTH_CLIENT_SECRET, source=SECRET_SOURCE_AZURE, resource_id=resource_id, resource_name=( f"sp/{c.application_display_name or c.application_id}" ), owner_person_id=None, owner_email=None, created_at_source=c.start_date, last_rotated_at=last_rotated, rotation_period_days=sp_period, next_rotation_due_at=next_due, status=status, days_until_due=days_until, raw_payload=c.raw, last_synced_at=synced_at, ) record_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation # Don't double-error if storage-keys already noted Azure # missing — but if SP-specific gap (Application.Read.All # scope) hits, surface that distinct cause. is_missing = isinstance(exc, IntegrationMissingError) if not is_missing: logger.warning( "auditor azure SP credentials sync failed: %s", exc, exc_info=True, ) errors.append({ "source": SECRET_SOURCE_AZURE, "error": ( str(exc) if is_missing else f"{type(exc).__name__}: {exc}" ), "kind": ( "service_principal_secrets" if is_missing else "upstream_error" ), }) await session.flush() from app.domains.auditor import findings_service as _findings # Per-source staleness gating (#2217): a provider that failed # this round is absent from ``scanned_sources``, so its rows are # exempt from the sweep and its HIGH findings stay open. await _findings.reconcile_secret_findings( session, scanned_sources=scanned_sources, ) overdue_count = (await session.execute( select(func.count()).select_from(AuditorSecretRotationRecord) .where(AuditorSecretRotationRecord.status == SECRET_STATUS_OVERDUE) )).scalar_one() await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="auditor.secret_rotation.sync", resource_type="auditor_secret_rotation_records", status="success", detail={ "scanned_sources": scanned_sources, "record_count": record_count, "overdue_count": int(overdue_count), "errors": errors, }, ) return { "scanned_sources": scanned_sources, "record_count": record_count, "overdue_count": int(overdue_count), "errors": errors, "synced_at": synced_at, } def _status_for_days(days_until: int | None) -> str: if days_until is None: return SECRET_STATUS_CURRENT if days_until < 0: return SECRET_STATUS_OVERDUE if days_until <= ROTATION_DUE_SOON_THRESHOLD_DAYS: return SECRET_STATUS_DUE_SOON return SECRET_STATUS_CURRENT async def _upsert_secret( session: AsyncSession, *, secret_kind: str, source: str, resource_id: str, resource_name: str | None, owner_person_id: uuid.UUID | None, owner_email: str | None, created_at_source: datetime | None, last_rotated_at: datetime | None, rotation_period_days: int, next_rotation_due_at: datetime | None, status: str, days_until_due: int | None, raw_payload: dict[str, Any] | None, last_synced_at: datetime, ) -> None: raw_payload = _json_safe(raw_payload) existing = ( await session.execute( select(AuditorSecretRotationRecord).where( AuditorSecretRotationRecord.source == source, AuditorSecretRotationRecord.secret_kind == secret_kind, AuditorSecretRotationRecord.resource_id == resource_id, ) ) ).scalar_one_or_none() if existing is None: session.add(AuditorSecretRotationRecord( id=uuid.uuid4(), secret_kind=secret_kind, source=source, resource_id=resource_id, resource_name=resource_name, owner_person_id=owner_person_id, owner_email=owner_email, created_at_source=created_at_source, last_rotated_at=last_rotated_at, rotation_period_days=rotation_period_days, next_rotation_due_at=next_rotation_due_at, status=status, days_until_due=days_until_due, raw_payload=raw_payload, last_synced_at=last_synced_at, )) return existing.resource_name = resource_name or existing.resource_name existing.owner_person_id = owner_person_id or existing.owner_person_id existing.owner_email = owner_email or existing.owner_email existing.created_at_source = created_at_source or existing.created_at_source existing.last_rotated_at = last_rotated_at or existing.last_rotated_at existing.rotation_period_days = rotation_period_days existing.next_rotation_due_at = next_rotation_due_at existing.status = status existing.days_until_due = days_until_due existing.raw_payload = raw_payload or existing.raw_payload existing.last_synced_at = last_synced_at # ═════════════════════════════════════════════════════════════════════ # 4. Signin anomaly monitoring # ═════════════════════════════════════════════════════════════════════ class SigninAnomalyService: """Pull recent signin events from each provider, compute anomaly score, persist into ``auditor_signin_events``. Scoring (additive, per event): * +30 if person.employment_status == 'left' (post-offboard signin) * +25 if event_country differs from the person's last 30d majority country (new country) * +15 if event_hour is outside [06:00, 23:59] in person's local time-zone (unusual hour) — TZ from person.timezone column or UTC fallback * +20 if 5+ failed signins in last 60 minutes from same person/IP * +35 if status == 'failure' and persons matches a leaver * +20 if user_agent matches a known headless client (curl, python-requests, etc.) Score is mapped to ``anomaly_level`` via ``constants.anomaly_level_for_score``. """ @staticmethod async def pull_and_score( session: AsyncSession, *, lookback_minutes: int = 70, actor_user_id: uuid.UUID | None = None, trace_id: uuid.UUID | None = None, ) -> dict[str, Any]: now = _utc_now() start_at = now - timedelta(minutes=lookback_minutes) pulled_providers: list[str] = [] new_event_count = 0 anomaly_count = 0 errors: list[dict[str, Any]] = [] emails = await _persons_by_email(session) # ── Pre-load anomaly context once per sync run ──────────── # 1. known_bad_ips list — admin-curated in platform_settings, # canonical key ``auditor.signin.known_bad_ips`` (JSON list # of strings). Empty list / missing setting = no IP marked # as known-bad — that is a *real* state, not a fallback. known_bad_ips = await _load_known_bad_ips(session) # 2. Per-person 30-day signin history for new-country baseline # + impossible_travel previous-coordinate. We pull every # person's last 30 days in one query, build an in-memory # index, and the per-event scoring just reads from it. history_window_start = now - timedelta(days=30) prior_signins_by_person = await _load_recent_signin_history( session, since=history_window_start, ) # 3. 60-minute sliding window of failures keyed by person_id # for rapid_failed_attempts. We pull failures only. failure_window_start = now - timedelta(minutes=60) failure_counts_by_person = await _load_recent_failure_counts( session, since=failure_window_start, ) # ─ Google Reports ───────────────────────────────────────── try: events = await GoogleAuditorAdapter.list_signin_events( session, start_at=start_at, end_at=now, ) pulled_providers.append(SIGNIN_PROVIDER_GOOGLE) except Exception as exc: # noqa: BLE001 — provider isolation events = [] _record_provider_error( errors, provider_label=SIGNIN_PROVIDER_GOOGLE, exc=exc, ) for ev in events: email_l = (ev.user_email or "").lower() person = emails.get(email_l) person_id = person.id if person else None geo = geoip_lookup(ev.source_ip) prior_history = ( prior_signins_by_person.get(person_id, []) if person_id else [] ) recent_failure_count = ( failure_counts_by_person.get(person_id, 0) if person_id else 0 ) # Failure events are bumped at write-time (after _score_event) # so we compute the count BEFORE adding self. score, reasons = _score_event( person=person, event_kind=ev.event_kind, status=ev.status, event_at=ev.event_at, source_ip=ev.source_ip, user_agent=ev.user_agent, geo=geo, prior_history=prior_history, recent_failure_count=recent_failure_count, known_bad_ips=known_bad_ips, ) level = anomaly_level_for_score(score) new_, was_high = await _upsert_signin( session, provider=SIGNIN_PROVIDER_GOOGLE, provider_event_id=ev.event_id, person_id=person_id, provider_user_id=None, provider_email=ev.user_email, event_at=ev.event_at, event_kind=ev.event_kind, status=ev.status, source_ip=ev.source_ip, source_country=geo.country if geo else None, source_city=geo.city if geo else None, user_agent=ev.user_agent, device_id=None, anomaly_score=score, anomaly_level=level, anomaly_reasons=reasons, raw_payload={ **ev.raw, **( {"_geo": { "lat": geo.latitude, "lon": geo.longitude, }} if geo and geo.latitude is not None else {} ), }, ) if new_: new_event_count += 1 # Maintain the in-memory failure-count and history so # the second event in the same batch sees the first. if person_id is not None: if ev.status == SIGNIN_STATUS_FAILURE: failure_counts_by_person[person_id] = ( failure_counts_by_person.get(person_id, 0) + 1 ) if geo and geo.latitude is not None: prior_signins_by_person.setdefault(person_id, []).append( _PriorSignin( event_at=ev.event_at, country=geo.country, latitude=geo.latitude, longitude=geo.longitude, ) ) if was_high: anomaly_count += 1 # ─ Feishu signin events ────────────────────────────────── try: f_events = await FeishuAuditorAdapter.list_signin_events( session, start_at=start_at, end_at=now, ) pulled_providers.append(SIGNIN_PROVIDER_FEISHU) except Exception as exc: # noqa: BLE001 — provider isolation f_events = [] _record_provider_error( errors, provider_label=SIGNIN_PROVIDER_FEISHU, exc=exc, ) for ev in f_events: email_l = (ev.user_email or "").lower() if ev.user_email else "" person = emails.get(email_l) if email_l else None person_id = person.id if person else None geo = geoip_lookup(ev.source_ip) prior_history = ( prior_signins_by_person.get(person_id, []) if person_id else [] ) recent_failure_count = ( failure_counts_by_person.get(person_id, 0) if person_id else 0 ) score, reasons = _score_event( person=person, event_kind=ev.event_kind, status=ev.status, event_at=ev.event_at, source_ip=ev.source_ip, user_agent=ev.user_agent, geo=geo, prior_history=prior_history, recent_failure_count=recent_failure_count, known_bad_ips=known_bad_ips, ) level = anomaly_level_for_score(score) new_, was_high = await _upsert_signin( session, provider=SIGNIN_PROVIDER_FEISHU, provider_event_id=ev.event_id, person_id=person_id, provider_user_id=ev.open_id, provider_email=ev.user_email, event_at=ev.event_at, event_kind=ev.event_kind, status=ev.status, source_ip=ev.source_ip, source_country=geo.country if geo else None, source_city=geo.city if geo else None, user_agent=ev.user_agent, device_id=None, anomaly_score=score, anomaly_level=level, anomaly_reasons=reasons, raw_payload={ **ev.raw, **( {"_geo": { "lat": geo.latitude, "lon": geo.longitude, }} if geo and geo.latitude is not None else {} ), }, ) if new_: new_event_count += 1 if person_id is not None: if ev.status == SIGNIN_STATUS_FAILURE: failure_counts_by_person[person_id] = ( failure_counts_by_person.get(person_id, 0) + 1 ) if geo and geo.latitude is not None: prior_signins_by_person.setdefault( person_id, [], ).append(_PriorSignin( event_at=ev.event_at, country=geo.country, latitude=geo.latitude, longitude=geo.longitude, )) if was_high: anomaly_count += 1 await session.flush() # Project high/critical unacknowledged anomalies into the worklist. from app.domains.auditor import findings_service as _findings await _findings.project_signin_findings(session) await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="auditor.signin.pull", resource_type="auditor_signin_events", status="success", detail={ "pulled_providers": pulled_providers, "new_event_count": new_event_count, "anomaly_count": anomaly_count, "lookback_minutes": lookback_minutes, "errors": errors, }, ) return { "pulled_providers": pulled_providers, "new_event_count": new_event_count, "anomaly_count": anomaly_count, "pulled_at": now, "errors": errors, } _HEADLESS_UA_PATTERNS = ( "curl/", "python-requests", "go-http-client", "okhttp", "aiohttp", "wget/", "java/", "axios/", "httpx/", "headless", ) # Threshold for ``rapid_failed_attempts`` — N failures within the # 60-minute pre-loaded window from the same person. RAPID_FAILED_ATTEMPTS_THRESHOLD = 5 # Speed threshold for ``impossible_travel``. If the inferred speed # (km / hours_between_events) exceeds this, geo-cheating is implied. # 900 km/h ≈ commercial-jet cruise; anything above is non-physical # (the upper-bound covers edge cases where two events differ by only # minutes — short Δt amplifies tiny GPS jitter to absurd speeds). IMPOSSIBLE_TRAVEL_SPEED_KMH = 900.0 # Minimum distance for impossible_travel — below this we treat it as # noise (same metro area, ISP-NAT round-trip, etc.). IMPOSSIBLE_TRAVEL_MIN_KM = 200.0 @dataclass(slots=True, frozen=True) class _PriorSignin: event_at: datetime country: str | None latitude: float | None longitude: float | None def _score_event( *, person: IdentityPerson | None, event_kind: str, status: str, event_at: datetime, source_ip: str | None, user_agent: str | None, geo: GeoResult | None, prior_history: list[_PriorSignin], recent_failure_count: int, known_bad_ips: frozenset[str], ) -> tuple[int, list[str]]: """Score a single signin event. All eight scoring rules now wired (was four) — the four newly activated rules (``impossible_travel`` / ``new_country`` / ``rapid_failed_attempts`` / ``known_bad_ip``) require the pre-loaded context the caller passes in. """ score = 0 reasons: list[str] = [] # Post-offboard signin (highest weight — leaver auth should be # impossible if leaver_blocked + force_logout did their jobs). # Direct attribute access — IdentityPerson always has # employment_status; a column rename surfaces as AttributeError # in tests instead of silently scoring 0. if person is not None and person.employment_status == "left": score += 30 reasons.append(ANOMALY_REASON_LOGIN_AFTER_OFFBOARD) # Plain failed event — small bump that compounds with rapid_attempts if status == SIGNIN_STATUS_FAILURE: score += 5 # Unusual hour (UTC heuristic — per-person TZ enrichment is a # follow-up; the prior_history hint already covers timezone # implicitly because most signins cluster in the work TZ band). # ``hour`` is 0-23, so the old ``hour > 23`` right branch was dead # code and the whole 23:00-23:59 band never scored. The intended # band is "outside 06:00-22:59 UTC". hour = event_at.astimezone(UTC).hour if hour < 6 or hour >= 23: score += 15 reasons.append(ANOMALY_REASON_UNUSUAL_HOUR) # Headless UA — automation / scripted abuse signal. Reported under # its own reason: labelling it ``new_device`` pointed the auditor at # device fingerprints instead of scripted access. if user_agent: ua_lower = user_agent.lower() if any(p in ua_lower for p in _HEADLESS_UA_PATTERNS): score += 20 reasons.append(ANOMALY_REASON_HEADLESS_UA) # Known bad IP — admin-curated blacklist. if source_ip and source_ip in known_bad_ips: score += 35 reasons.append(ANOMALY_REASON_KNOWN_BAD_IP) # New country — compare event country to the person's last 30d # majority country. Skip when geo unavailable OR person has < 3 # prior events (insufficient baseline to call any country "new"). if geo and geo.country and len(prior_history) >= 3: prior_countries = [ h.country for h in prior_history if h.country ] if prior_countries: from collections import Counter top_country, top_count = Counter(prior_countries).most_common(1)[0] # Treat as "new" only if the prior majority is decisive # (≥ 60% of the prior baseline) AND the current country # differs. Avoids triggering on a person whose history is # already evenly split between two countries. if ( top_country and top_country != geo.country and top_count / len(prior_countries) >= 0.6 ): score += 25 reasons.append(ANOMALY_REASON_NEW_COUNTRY) # Impossible travel — distance/time vs. last prior event with geo. if ( geo and geo.latitude is not None and geo.longitude is not None and prior_history ): # Find the closest-in-time prior event with geo coords. prior_with_geo = [ h for h in prior_history if h.latitude is not None and h.longitude is not None and h.event_at < event_at ] if prior_with_geo: most_recent = max(prior_with_geo, key=lambda h: h.event_at) dt_hours = (event_at - most_recent.event_at).total_seconds() / 3600.0 if dt_hours > 0: dist_km = haversine_km( most_recent.latitude, most_recent.longitude, geo.latitude, geo.longitude, ) if dist_km >= IMPOSSIBLE_TRAVEL_MIN_KM: speed = dist_km / dt_hours if speed > IMPOSSIBLE_TRAVEL_SPEED_KMH: score += 30 reasons.append(ANOMALY_REASON_IMPOSSIBLE_TRAVEL) # Rapid failed attempts — N+ failures from same person in 60min. if ( recent_failure_count >= RAPID_FAILED_ATTEMPTS_THRESHOLD and ANOMALY_REASON_RAPID_FAILED_ATTEMPTS not in reasons ): score += 20 reasons.append(ANOMALY_REASON_RAPID_FAILED_ATTEMPTS) return score, reasons # ───────────────────────────────────────────────────────────────────── # Anomaly context loaders # ───────────────────────────────────────────────────────────────────── async def _load_known_bad_ips(session: AsyncSession) -> frozenset[str]: """Load admin-curated bad-IP list from platform_settings. Key: ``auditor.signin.known_bad_ips`` — a JSON list of IP literals (IPv4 or IPv6 strings, exact match). Whitespace is stripped per entry; duplicates dropped. Missing / empty / non-list values yield an empty set — that is the *real* configured state, never a fake default. """ from app.integrations.models import PlatformSetting row = (await session.execute( select(PlatformSetting).where( PlatformSetting.key == "auditor.signin.known_bad_ips", ) )).scalar_one_or_none() if row is None or row.value is None: return frozenset() raw = row.value if isinstance(raw, str): try: import json as _json raw = _json.loads(raw) except (ValueError, TypeError): return frozenset() if not isinstance(raw, list): return frozenset() return frozenset( s.strip() for s in raw if isinstance(s, str) and s.strip() ) async def _load_recent_signin_history( session: AsyncSession, *, since: datetime, ) -> dict[uuid.UUID, list[_PriorSignin]]: """Pre-load signin events newer than ``since``, grouped by person. Used by ``_score_event`` for ``new_country`` and ``impossible_travel`` reasons. Caps each person's history at the most recent 50 events to keep memory bounded; in real fleets a person sees ~20-30 events / month so the cap rarely triggers. """ rows = (await session.execute( select( AuditorSigninEvent.person_id, AuditorSigninEvent.event_at, AuditorSigninEvent.source_country, AuditorSigninEvent.raw_payload, ).where( AuditorSigninEvent.event_at >= since, AuditorSigninEvent.person_id.is_not(None), ).order_by(AuditorSigninEvent.event_at.desc()) )).all() out: dict[uuid.UUID, list[_PriorSignin]] = {} for pid, event_at, country, raw_payload in rows: bucket = out.setdefault(pid, []) if len(bucket) >= 50: continue # ``raw_payload._geo`` is set by this service when a fresh # event is upserted with geo. Older events that pre-date the # geoip plumbing won't have it; treat lat/lon as unavailable. lat: float | None = None lon: float | None = None if isinstance(raw_payload, dict): geo_blob = raw_payload.get("_geo") if isinstance(geo_blob, dict): lat_v = geo_blob.get("lat") lon_v = geo_blob.get("lon") if isinstance(lat_v, (int, float)): lat = float(lat_v) if isinstance(lon_v, (int, float)): lon = float(lon_v) bucket.append(_PriorSignin( event_at=event_at, country=country, latitude=lat, longitude=lon, )) return out async def _load_recent_failure_counts( session: AsyncSession, *, since: datetime, ) -> dict[uuid.UUID, int]: """Pre-load failure-event counts newer than ``since`` per person. Used by ``_score_event`` for the ``rapid_failed_attempts`` reason. A small (count) integer per person; the whole map fits in memory even for tens of thousands of users. """ rows = (await session.execute( select( AuditorSigninEvent.person_id, func.count(AuditorSigninEvent.id), ).where( AuditorSigninEvent.event_at >= since, AuditorSigninEvent.status == SIGNIN_STATUS_FAILURE, AuditorSigninEvent.person_id.is_not(None), ).group_by(AuditorSigninEvent.person_id) )).all() return {pid: int(c) for pid, c in rows} async def _upsert_signin( session: AsyncSession, *, provider: str, provider_event_id: str | None, person_id: uuid.UUID | None, provider_user_id: str | None, provider_email: str | None, event_at: datetime, event_kind: str, status: str, source_ip: str | None, source_country: str | None, source_city: str | None, user_agent: str | None, device_id: str | None, anomaly_score: int, anomaly_level: str, anomaly_reasons: list[str], raw_payload: dict[str, Any] | None, ) -> tuple[bool, bool]: raw_payload = _json_safe(raw_payload) if provider_event_id: existing = (await session.execute( select(AuditorSigninEvent).where( AuditorSigninEvent.provider == provider, AuditorSigninEvent.provider_event_id == provider_event_id, ) )).scalar_one_or_none() if existing is not None: return False, anomaly_level in (ANOMALY_LEVEL_HIGH, ANOMALY_LEVEL_CRITICAL) session.add(AuditorSigninEvent( id=uuid.uuid4(), person_id=person_id, provider=provider, provider_event_id=provider_event_id, provider_user_id=provider_user_id, provider_email=provider_email, event_at=event_at, event_kind=event_kind, status=status, source_ip=source_ip, source_country=source_country, source_city=source_city, user_agent=user_agent, device_id=device_id, anomaly_score=anomaly_score, anomaly_level=anomaly_level, anomaly_reasons=anomaly_reasons, raw_payload=raw_payload, )) return True, anomaly_level in (ANOMALY_LEVEL_HIGH, ANOMALY_LEVEL_CRITICAL) # ═════════════════════════════════════════════════════════════════════ # 5. PAT audit # ═════════════════════════════════════════════════════════════════════ class PatAuditService: """Refresh ``auditor_pat_audits`` from: * GitHub org PATs (Enterprise admin endpoint) * Helix's own ``apikey_keys`` (treated as MCP-shaped tokens) * Helix ``auth_external_identities`` rows where provider has OAuth grants (Google / Feishu) — surfaced as audit rows but NOT revocable from here. Risk flags computed on each sync: * ``no_expiry`` — token has no expiry date * ``dormant`` — last_used > 90d ago * ``over_scoped`` / ``broad_admin_scope`` — token holds admin scopes * ``owner_left`` — owner.employment_status='left' """ @staticmethod async def sync( session: AsyncSession, *, actor_user_id: uuid.UUID | None = None, trace_id: uuid.UUID | None = None, ) -> dict[str, Any]: synced_at = _utc_now() pulled_sources: list[str] = [] record_count = 0 risky_count = 0 errors: list[dict[str, Any]] = [] emails = await _persons_by_email(session) person_ids_left = { p.id for p in emails.values() if p.employment_status == "left" } # ─ GitHub org PATs ──────────────────────────────────────── try: gh_pats = await GitHubAuditorAdapter.list_organization_pats(session) pulled_sources.append(PAT_SOURCE_GITHUB) except Exception as exc: # noqa: BLE001 — provider isolation gh_pats = [] _record_provider_error( errors, source_label=PAT_SOURCE_GITHUB, exc=exc, ) emails_by_local_pat = _by_local_part(emails) for p in gh_pats: login = (p.owner_login or "").lower() person = emails_by_local_pat.get(login) risk_flags = _pat_risk_flags( scopes=p.scopes, expires_at=p.expires_at, last_used_at=p.last_used_at, synced_at=synced_at, owner_left=(person.id in person_ids_left) if person else False, revoked=bool(p.is_revoked), ) status = ( PAT_STATUS_REVOKED if p.is_revoked else PAT_STATUS_EXPIRED if (p.expires_at and p.expires_at < synced_at) else PAT_STATUS_DORMANT if PAT_RISK_DORMANT in risk_flags else PAT_STATUS_ACTIVE ) await _upsert_pat( session, kind=PAT_KIND_GITHUB_PAT, source=PAT_SOURCE_GITHUB, token_external_id=p.token_id, person_id=person.id if person else None, owner_email=person.work_email if person else None, token_name=p.name, scopes=p.scopes, created_at_source=p.created_at, last_used_at=p.last_used_at, expires_at=p.expires_at, status=status, risk_flags=risk_flags, raw_payload=p.raw, last_synced_at=synced_at, ) record_count += 1 if risk_flags: risky_count += 1 # ─ Helix API key assignments (treat as MCP-grade tokens) ── try: from app.domains.apikey.models import ApiKeyAssignment apikey_rows = (await session.execute( select(ApiKeyAssignment) )).scalars().all() pulled_sources.append(PAT_SOURCE_HELIX_APIKEY) for k in apikey_rows: # ApiKeyAssignment carries valid_until (date) as expiry. expires_at = ( datetime(k.valid_until.year, k.valid_until.month, k.valid_until.day, tzinfo=UTC) if k.valid_until else None ) last_used = None # No usage tracking on this table # No explicit scopes column — provider_id maps to one # ApiKeyProvider; we surface its name as the single # scope for risk-flag purposes. scopes: list[str] = [] risk_flags = _pat_risk_flags( scopes=scopes, expires_at=expires_at, last_used_at=last_used, synced_at=synced_at, owner_left=(k.assigned_to in person_ids_left), revoked=k.status in ("revoked", "rejected", "cancelled"), ) status = ( PAT_STATUS_REVOKED if k.status in ( "revoked", "rejected", "cancelled", ) else PAT_STATUS_EXPIRED if ( expires_at and expires_at < synced_at ) else PAT_STATUS_DORMANT if PAT_RISK_DORMANT in risk_flags else PAT_STATUS_ACTIVE ) await _upsert_pat( session, kind=PAT_KIND_HELIX_API_KEY, source=PAT_SOURCE_HELIX_APIKEY, token_external_id=str(k.id), person_id=k.assigned_to, owner_email=None, token_name=k.project_name, scopes=scopes, created_at_source=_aware(k.created_at), last_used_at=last_used, expires_at=expires_at, status=status, risk_flags=risk_flags, raw_payload={ "request_number": k.request_number, "id": str(k.id), "project_name": k.project_name, "status": k.status, }, last_synced_at=synced_at, ) record_count += 1 if risk_flags: risky_count += 1 except ImportError: errors.append({ "source": PAT_SOURCE_HELIX_APIKEY, "error": "apikey domain not installed", }) # ─ Helix PAT (auth_personal_access_tokens) ───────────────── # The native Helix PAT table is the most important PAT data # source for this org — every AI client (Claude Desktop / # Cursor / Helix MCP) authenticates via these tokens, and the # table carries real ``last_used_at`` / ``last_used_ip`` / # ``revoked_at`` columns that the GitHub-PAT path does not. try: from app.auth.models import AuthRole, AuthUserRole from app.auth.tokens.models import PersonalAccessToken # Build (person_id → set[role_name]) without depending on # the lazy-loaded IdentityPerson.role_names property — async # SQLAlchemy raises MissingGreenlet on lazy relationship # access. roles_by_person: dict[uuid.UUID, set[str]] = defaultdict(set) ur_rows = (await session.execute( select(AuthUserRole.user_id, AuthRole.name) .join(AuthRole, AuthRole.id == AuthUserRole.role_id) )).all() for pid, role_name in ur_rows: roles_by_person[pid].add(role_name) emails_by_pid = {p.id: p for p in emails.values()} pat_rows = (await session.execute( select(PersonalAccessToken) )).scalars().all() pulled_sources.append(PAT_SOURCE_HELIX_PAT) for t in pat_rows: # Helix PATs inherit the owner's full role set — there # is no per-token scope. We treat any privileged role # holder's PAT as broad-admin scope so the risk score # and dashboard surface PATs owned by admins as elevated # risk. owner = emails_by_pid.get(t.user_id) synthetic_scopes: list[str] = [] role_set = roles_by_person.get(t.user_id, set()) if role_set & { "global_admin", "it_admin", "security_auditor", "asset_admin", "cloud_admin", "hr", }: # Map privileged Helix role to a GitHub-shaped admin # scope so the existing _pat_risk_flags helper # catches it (PAT_BROAD_ADMIN_SCOPES contains # "admin:org"). synthetic_scopes.append("admin:org") # Coerce all DB-sourced datetimes to tz-aware before # arithmetic — MySQL TIMESTAMP returns naive even # though the column is declared timezone=True. t_expires_at = _aware(t.expires_at) t_last_used_at = _aware(t.last_used_at) t_revoked_at = _aware(t.revoked_at) t_created_at = _aware(t.created_at) risk_flags = _pat_risk_flags( scopes=synthetic_scopes, expires_at=t_expires_at, last_used_at=t_last_used_at, synced_at=synced_at, owner_left=(t.user_id in person_ids_left), revoked=t_revoked_at is not None, ) if t_revoked_at is not None: status = PAT_STATUS_REVOKED elif t_expires_at is not None and t_expires_at < synced_at: status = PAT_STATUS_EXPIRED elif PAT_RISK_DORMANT in risk_flags: status = PAT_STATUS_DORMANT else: status = PAT_STATUS_ACTIVE await _upsert_pat( session, kind=PAT_KIND_HELIX_PAT, source=PAT_SOURCE_HELIX_PAT, token_external_id=str(t.id), person_id=t.user_id, owner_email=(owner.work_email if owner else None), token_name=t.name, scopes=synthetic_scopes, created_at_source=t_created_at, last_used_at=t_last_used_at, expires_at=t_expires_at, status=status, risk_flags=risk_flags, raw_payload={ "id": str(t.id), "token_prefix": t.token_prefix, "last_used_ip": t.last_used_ip, "last_used_user_agent": ( t.last_used_user_agent[:200] if t.last_used_user_agent else None ), "revoked_at": ( t_revoked_at.isoformat() if t_revoked_at else None ), }, last_synced_at=synced_at, ) record_count += 1 if risk_flags: risky_count += 1 except ImportError: errors.append({ "source": PAT_SOURCE_HELIX_PAT, "error": "auth.tokens module not installed", }) # ─ Atlassian organization API tokens ────────────────────── try: at_tokens = await AtlassianAuditorAdapter.list_organization_api_tokens( session, ) pulled_sources.append(PAT_SOURCE_ATLASSIAN) for t in at_tokens: # Match holder by email when present. person = ( emails.get(t.user_email.lower()) if t.user_email else None ) # Atlassian Org API tokens have no scope list on the # public API surface — they implicitly grant org-admin # power, which is the broadest possible. Tag every one # as broad_admin by default so the risk score reflects. synthetic_scopes = ["admin:org"] risk_flags = _pat_risk_flags( scopes=synthetic_scopes, expires_at=None, # no expiry on these last_used_at=t.last_used_at, synced_at=synced_at, owner_left=( person.id in person_ids_left if person else False ), # Atlassian's listing endpoint exposes no # revocation state — see the note just below. revoked=False, ) # Atlassian API tokens have no public revocation/expiry # state on the listing endpoint; treat as active. The # auditor surfaces them as ``no_expiry`` + dormant if # unused — the operator action is to delete from the # org admin UI. if PAT_RISK_DORMANT in risk_flags: status = PAT_STATUS_DORMANT else: status = PAT_STATUS_ACTIVE await _upsert_pat( session, kind=PAT_KIND_ATLASSIAN_API_TOKEN, source=PAT_SOURCE_ATLASSIAN, token_external_id=t.token_id, person_id=(person.id if person else None), owner_email=t.user_email, token_name=t.label, scopes=synthetic_scopes, created_at_source=t.created_at, last_used_at=t.last_used_at, expires_at=None, status=status, risk_flags=risk_flags, raw_payload=t.raw, last_synced_at=synced_at, ) record_count += 1 if risk_flags: risky_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=PAT_SOURCE_ATLASSIAN, exc=exc, ) # ─ Helix OAuth grants (auth_external_identities) ─────────── try: from app.identity.person_models import IdentityExternalIdentity grants = (await session.execute( select(IdentityExternalIdentity).where( IdentityExternalIdentity.provider != "successfactors", IdentityExternalIdentity.deleted_at.is_(None), ) )).scalars().all() pulled_sources.append(PAT_SOURCE_HELIX_OAUTH) for g in grants: # OAuth grant has no expiry/scope info on the row. # Surface as audit-only. risk_flags = ( [PAT_RISK_OWNER_LEFT] if g.person_id in person_ids_left else [] ) status = PAT_STATUS_ACTIVE await _upsert_pat( session, kind=PAT_KIND_OAUTH_GRANT, source=PAT_SOURCE_HELIX_OAUTH, token_external_id=f"{g.provider}:{g.provider_subject}", person_id=g.person_id, owner_email=g.provider_email, token_name=f"oauth:{g.provider}", scopes=None, created_at_source=_aware(g.created_at), last_used_at=_aware(g.last_synced_at), expires_at=None, status=status, risk_flags=risk_flags, raw_payload={ "provider": g.provider, "subject": g.provider_subject, "username": g.provider_username, }, last_synced_at=synced_at, ) record_count += 1 if risk_flags: risky_count += 1 except ImportError: errors.append({ "source": PAT_SOURCE_HELIX_OAUTH, "error": "identity domain not installed", }) await session.flush() from app.domains.auditor import findings_service as _findings await _findings.reconcile_pat_findings( session, scanned_sources=pulled_sources, ) await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="auditor.pat.sync", resource_type="auditor_pat_audits", status="success", detail={ "pulled_sources": pulled_sources, "record_count": record_count, "risky_count": risky_count, "errors": errors, }, ) return { "pulled_sources": pulled_sources, "record_count": record_count, "risky_count": risky_count, "synced_at": synced_at, "errors": errors, } def _pat_risk_flags( *, scopes: list[str] | None, expires_at: datetime | None, last_used_at: datetime | None, synced_at: datetime, owner_left: bool, revoked: bool = False, ) -> list[str]: """Risk flags for one token. A REVOKED token carries none (#2137). Without this a revoked Helix PAT keeps its row, and 90 days later it trivially satisfies ``dormant`` — so it gets RE-flagged and counted into ``risky_count`` and into a "Risky token" finding, for a credential that no longer exists. Dormancy is a statement about a LIVE token; a dead one is not dormant, it is dead. """ if revoked: return [] flags: list[str] = [] if expires_at is None: flags.append(PAT_RISK_NO_EXPIRY) scopes_set = set(scopes or []) if scopes_set & PAT_BROAD_ADMIN_SCOPES: flags.append(PAT_RISK_BROAD_SCOPE_ADMIN) flags.append(PAT_RISK_OVER_SCOPED) if last_used_at is None or ( synced_at - last_used_at >= timedelta(days=PAT_DORMANT_DAYS) ): flags.append(PAT_RISK_DORMANT) if owner_left: flags.append(PAT_RISK_OWNER_LEFT) return flags async def _upsert_pat( session: AsyncSession, *, kind: str, source: str, token_external_id: str, person_id: uuid.UUID | None, owner_email: str | None, token_name: str | None, scopes: list[str] | None, created_at_source: datetime | None, last_used_at: datetime | None, expires_at: datetime | None, status: str, risk_flags: list[str], raw_payload: dict[str, Any] | None, last_synced_at: datetime, ) -> None: raw_payload = _json_safe(raw_payload) existing = (await session.execute( select(AuditorPatAudit).where( AuditorPatAudit.source == source, AuditorPatAudit.kind == kind, AuditorPatAudit.token_external_id == token_external_id, ) )).scalar_one_or_none() if existing is None: session.add(AuditorPatAudit( id=uuid.uuid4(), kind=kind, source=source, token_external_id=token_external_id, person_id=person_id, owner_email=owner_email, token_name=token_name, scopes=scopes, created_at_source=created_at_source, last_used_at=last_used_at, expires_at=expires_at, status=status, risk_flags=risk_flags, raw_payload=raw_payload, last_synced_at=last_synced_at, )) return existing.person_id = person_id or existing.person_id existing.owner_email = owner_email or existing.owner_email existing.token_name = token_name or existing.token_name existing.scopes = scopes existing.created_at_source = created_at_source or existing.created_at_source existing.last_used_at = last_used_at or existing.last_used_at existing.expires_at = expires_at or existing.expires_at existing.status = status existing.risk_flags = risk_flags existing.raw_payload = raw_payload or existing.raw_payload existing.last_synced_at = last_synced_at # ═════════════════════════════════════════════════════════════════════ # 6. Cert + ICP expiry sync # ═════════════════════════════════════════════════════════════════════ class CertExpirySyncService: """Refresh ``auditor_cert_expiries`` from AWS ACM + manual entries. Manual entries (ICP filings, off-AWS SSL certs, .com/.cn domain expirations) are upserted via the router endpoint ``POST /auditor/cert-expiries/manual`` — this sync only refreshes the ``alert_level`` and ``status`` columns based on ``expires_at``. AWS ACM rows are full upserts. """ @staticmethod async def sync( session: AsyncSession, *, actor_user_id: uuid.UUID | None = None, trace_id: uuid.UUID | None = None, ) -> dict[str, Any]: synced_at = _utc_now() pulled_sources: list[str] = [] record_count = 0 alerting_count = 0 errors: list[dict[str, Any]] = [] # ─ AWS ACM ──────────────────────────────────────────────── try: certs = await AwsAuditorAdapter.list_acm_certs(session) pulled_sources.append(CERT_SOURCE_AWS_ACM) for c in certs: if c.not_after is None: continue days_until = int( (c.not_after - synced_at).total_seconds() // 86400 ) alert = cert_alert_level_for_days(days_until) status = ( CERT_STATUS_EXPIRED if days_until < 0 else CERT_STATUS_ACTIVE ) await _upsert_cert( session, asset_kind=CERT_KIND_SSL, source=CERT_SOURCE_AWS_ACM, external_id=c.arn, name=c.domain_name, subject=c.domain_name, issuer=c.issuer, issued_at=c.not_before, expires_at=c.not_after, days_until_expiry=days_until, alert_level=alert, status=status, owner_person_id=None, owner_email=None, environment=None, notes=None, raw_payload=c.raw, last_synced_at=synced_at, ) record_count += 1 if alert != "none": alerting_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=CERT_SOURCE_AWS_ACM, exc=exc, ) # ─ Aliyun registered domains ─────────────────────────────── try: aly_domains = await AliyunAuditorAdapter.list_domains(session) pulled_sources.append(CERT_SOURCE_ALIYUN_DOMAIN) for d in aly_domains: if d.expiration_date is None: continue days_until = int( (d.expiration_date - synced_at).total_seconds() // 86400 ) alert = cert_alert_level_for_days(days_until) status = ( CERT_STATUS_EXPIRED if days_until < 0 else CERT_STATUS_ACTIVE ) await _upsert_cert( session, asset_kind=CERT_KIND_DOMAIN, source=CERT_SOURCE_ALIYUN_DOMAIN, external_id=d.domain_name, name=d.domain_name, subject=d.domain_name, issuer=None, issued_at=d.registration_date, expires_at=d.expiration_date, days_until_expiry=days_until, alert_level=alert, status=status, owner_person_id=None, owner_email=None, environment=None, notes=( f"ICP: {d.icp_filing_no}" if d.icp_filing_no else None ), raw_payload=d.raw, last_synced_at=synced_at, ) record_count += 1 if alert != "none": alerting_count += 1 # Domains carrying an ICP filing number get a sibling # row of asset_kind=ICP so the dashboard's ICP filter # picks them up. ICP filings don't expire on their own # in China, but they DO get revoked when the domain # expires — so we mirror the domain's expiry as the # alert anchor. if d.icp_filing_no: await _upsert_cert( session, asset_kind=CERT_KIND_ICP, source=CERT_SOURCE_ALIYUN_DOMAIN, external_id=f"icp:{d.domain_name}", name=d.domain_name, subject=d.icp_filing_no, issuer="MIIT (via Aliyun)", issued_at=d.registration_date, expires_at=d.expiration_date, days_until_expiry=days_until, alert_level=alert, status=status, owner_person_id=None, owner_email=None, environment=None, notes=d.icp_filing_status, raw_payload=d.raw, last_synced_at=synced_at, ) record_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=CERT_SOURCE_ALIYUN_DOMAIN, exc=exc, ) # ─ Tencent registered domains ────────────────────────────── try: tc_domains = await TencentAuditorAdapter.list_domains(session) pulled_sources.append(CERT_SOURCE_TENCENT_DOMAIN) for d in tc_domains: if d.expiration_date is None: continue days_until = int( (d.expiration_date - synced_at).total_seconds() // 86400 ) alert = cert_alert_level_for_days(days_until) status = ( CERT_STATUS_EXPIRED if days_until < 0 else CERT_STATUS_ACTIVE ) await _upsert_cert( session, asset_kind=CERT_KIND_DOMAIN, source=CERT_SOURCE_TENCENT_DOMAIN, external_id=d.domain_name, name=d.domain_name, subject=d.domain_name, issuer=None, issued_at=d.registration_date, expires_at=d.expiration_date, days_until_expiry=days_until, alert_level=alert, status=status, owner_person_id=None, owner_email=None, environment=None, notes=None, raw_payload=d.raw, last_synced_at=synced_at, ) record_count += 1 if alert != "none": alerting_count += 1 except Exception as exc: # noqa: BLE001 — provider isolation _record_provider_error( errors, source_label=CERT_SOURCE_TENCENT_DOMAIN, exc=exc, ) # ─ Manual entries — refresh alert / status only ──────────── manual_rows = (await session.execute( select(AuditorCertExpiry).where( AuditorCertExpiry.source == CERT_SOURCE_MANUAL, ) )).scalars().all() if manual_rows: pulled_sources.append(CERT_SOURCE_MANUAL) for r in manual_rows: if r.expires_at is None: continue # ``r.expires_at`` is freshly DB-loaded → tz-naive even though # the column is ``DateTime(timezone=True)``; ``synced_at`` is # aware. Subtracting them directly raises TypeError and kills # the whole cert/domain expiry sync run. Coerce via ``_aware`` # like every other DB-sourced operand in this function. days_until = int( (_aware(r.expires_at) - synced_at).total_seconds() // 86400 ) r.days_until_expiry = days_until new_alert = cert_alert_level_for_days(days_until) new_status = ( CERT_STATUS_EXPIRED if days_until < 0 else CERT_STATUS_ACTIVE if r.status not in (CERT_STATUS_EXPIRED,) else r.status ) r.alert_level = new_alert r.status = new_status r.last_synced_at = synced_at record_count += 1 if new_alert != "none": alerting_count += 1 await session.flush() from app.domains.auditor import findings_service as _findings await _findings.reconcile_cert_findings( session, scanned_sources=pulled_sources, ) await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="auditor.cert_expiry.sync", resource_type="auditor_cert_expiries", status="success", detail={ "pulled_sources": pulled_sources, "record_count": record_count, "alerting_count": alerting_count, "errors": errors, }, ) return { "pulled_sources": pulled_sources, "record_count": record_count, "alerting_count": alerting_count, "synced_at": synced_at, "errors": errors, } async def _upsert_cert( session: AsyncSession, *, asset_kind: str, source: str, external_id: str, name: str, subject: str | None, issuer: str | None, issued_at: datetime | None, expires_at: datetime, days_until_expiry: int | None, alert_level: str, status: str, owner_person_id: uuid.UUID | None, owner_email: str | None, environment: str | None, notes: str | None, raw_payload: dict[str, Any] | None, last_synced_at: datetime, ) -> None: raw_payload = _json_safe(raw_payload) existing = (await session.execute( select(AuditorCertExpiry).where( AuditorCertExpiry.source == source, AuditorCertExpiry.asset_kind == asset_kind, AuditorCertExpiry.external_id == external_id, ) )).scalar_one_or_none() if existing is None: session.add(AuditorCertExpiry( id=uuid.uuid4(), asset_kind=asset_kind, source=source, external_id=external_id, name=name, subject=subject, issuer=issuer, issued_at=issued_at, expires_at=expires_at, days_until_expiry=days_until_expiry, alert_level=alert_level, status=status, owner_person_id=owner_person_id, owner_email=owner_email, environment=environment, notes=notes, raw_payload=raw_payload, last_synced_at=last_synced_at, )) return existing.name = name existing.subject = subject or existing.subject existing.issuer = issuer or existing.issuer existing.issued_at = issued_at or existing.issued_at existing.expires_at = expires_at existing.days_until_expiry = days_until_expiry existing.alert_level = alert_level existing.status = status existing.owner_person_id = owner_person_id or existing.owner_person_id existing.owner_email = owner_email or existing.owner_email existing.environment = environment or existing.environment existing.notes = notes if notes is not None else existing.notes existing.raw_payload = raw_payload or existing.raw_payload existing.last_synced_at = last_synced_at