"""System registry CRUD, ownership transfer, and the reads the rest of the platform asks for. Two rules shape this module: * **Ownership writes are events.** Every owner change appends a ``RegisteredSystemEvent`` so "who was responsible in May" is answerable after the fact. An incident review that has to guess is the failure this table exists to prevent. * **Hostname matching suggests, never decides.** ``suggest_for_host`` returns candidates for a human to accept. Auto-linking a report to a system would silently re-assign the finding's owner and budget pot on a substring match. """ from __future__ import annotations import re import uuid from datetime import UTC, datetime from typing import Any from sqlalchemy import Select, Text, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import aliased, selectinload from app.core.exceptions import ConflictError, NotFoundError, ValidationError from app.domains.system_registry import constants as const from app.domains.system_registry.models import ( RegisteredSystem, RegisteredSystemEvent, ) from app.identity.person_models import IdentityPerson _SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,62}[a-z0-9]$") _HOST_RE = re.compile(r"^[a-z0-9]([a-z0-9.-]{0,251}[a-z0-9])?$") def _now() -> datetime: return datetime.now(UTC) def _naive(dt: datetime) -> datetime: return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt # ════════════════════════════════════════════════════════════════════ # Validation # ════════════════════════════════════════════════════════════════════ def normalize_slug(raw: str) -> str: slug = (raw or "").strip().lower().replace("_", "-") slug = re.sub(r"-{2,}", "-", slug) if not _SLUG_RE.match(slug): raise ValidationError( "slug must be lower-case letters, digits and hyphens " "(2-64 chars), e.g. 'tidb-cloud'", ) return slug def normalize_hostnames(raw: list[str] | None) -> list[str]: """Lower-case, de-duplicate, strip scheme/path, and bound the list. Accepting ``https://clinic.pingcap.com/health`` and storing the bare host is deliberate: operators paste URLs out of the researcher's mail, and a stored value with a scheme would never match the hostname a later report names.""" if not raw: return [] out: list[str] = [] for item in raw: host = (item or "").strip().lower() if not host: continue host = re.sub(r"^[a-z]+://", "", host) host = host.split("/", 1)[0].split("?", 1)[0] host = host.split(":", 1)[0].rstrip(".") if not host: continue if not _HOST_RE.match(host): raise ValidationError(f"'{item}' is not a valid hostname") if host not in out: out.append(host) if len(out) > const.MAX_HOSTS: raise ValidationError( f"at most {const.MAX_HOSTS} hostnames per system", ) return out def _validate_enum(value: str, allowed: tuple[str, ...], field: str) -> str: if value not in allowed: raise ValidationError( f"{field} must be one of: {', '.join(allowed)}", ) return value def _bounded(value: str | None, limit: int, field: str) -> str | None: if value is None: return None text = value.strip() if not text: return None if len(text) > limit: raise ValidationError(f"{field} exceeds {limit} characters") return text # ════════════════════════════════════════════════════════════════════ # Reads # ════════════════════════════════════════════════════════════════════ def _base_query() -> Select[tuple[RegisteredSystem]]: return ( select(RegisteredSystem) .where(RegisteredSystem.deleted_at.is_(None)) .options( selectinload(RegisteredSystem.owner), selectinload(RegisteredSystem.backup_owner), ) # ``populate_existing`` is required, not defensive. The session # runs with ``expire_on_commit=False``, so a row already in the # identity map keeps whatever ``owner`` was loaded earlier — # re-reading after ``transfer_owner`` would hand back the PREVIOUS # owner (or ``None``) while the column in the database is # correct. Caught by the transfer test, not by inspection. .execution_options(populate_existing=True) ) async def get_system(db: AsyncSession, system_id: uuid.UUID) -> RegisteredSystem: row = ( await db.execute(_base_query().where(RegisteredSystem.id == system_id)) ).scalar_one_or_none() if row is None: raise NotFoundError("System", str(system_id)) return row async def get_by_slug(db: AsyncSession, slug: str) -> RegisteredSystem | None: return ( await db.execute( _base_query().where(RegisteredSystem.slug == normalize_slug(slug)), ) ).scalar_one_or_none() async def list_systems( db: AsyncSession, *, q: str | None = None, status: str | None = None, kind: str | None = None, exposure: str | None = None, owner_person_id: uuid.UUID | None = None, ownerless: bool | None = None, in_bounty_scope: bool | None = None, budget_track: str | None = None, page: int = 1, page_size: int = 20, ) -> tuple[list[RegisteredSystem], int]: stmt = _base_query() if q: needle = f"%{q.strip().lower()}%" stmt = stmt.where( or_( func.lower(RegisteredSystem.name).like(needle), func.lower(RegisteredSystem.slug).like(needle), func.lower( func.cast(RegisteredSystem.hostnames, Text), ).like(needle), ), ) if status: stmt = stmt.where(RegisteredSystem.status == status) if kind: stmt = stmt.where(RegisteredSystem.kind == kind) if exposure: stmt = stmt.where(RegisteredSystem.exposure == exposure) if owner_person_id is not None: stmt = stmt.where( or_( RegisteredSystem.owner_person_id == owner_person_id, RegisteredSystem.backup_owner_person_id == owner_person_id, ), ) if ownerless is not None: # "Ownerless" on the wire means "nobody is actually answering for # it", which includes a named owner who has left — the KPI tile # and this filter must select the same rows, or clicking the tile # shows fewer systems than the number on it. gone = aliased(IdentityPerson) stmt = stmt.outerjoin( gone, gone.id == RegisteredSystem.owner_person_id, ) unowned = or_( RegisteredSystem.owner_person_id.is_(None), gone.employment_status != "active", gone.deleted_at.is_not(None), ) stmt = stmt.where(unowned if ownerless else ~unowned) if in_bounty_scope is not None: stmt = stmt.where(RegisteredSystem.in_bounty_scope.is_(in_bounty_scope)) if budget_track: stmt = stmt.where(RegisteredSystem.budget_track == budget_track) total = ( await db.execute( select(func.count()).select_from(stmt.subquery()), ) ).scalar_one() rows = ( await db.execute( # ``status`` ascending sorts active → deprecated → retired # (alphabetical, and deliberately the order an operator wants: # what is running first, what is gone last). stmt.order_by( RegisteredSystem.status.asc(), RegisteredSystem.name.asc(), ) .offset((page - 1) * page_size) .limit(page_size), ) ).scalars().all() return list(rows), int(total) async def suggest_for_host( db: AsyncSession, needle: str, *, limit: int = 5, ) -> list[RegisteredSystem]: """Candidate systems for a free-text "affected system" string. Matches on hostname substring first, then on name. Returns a list for a human to pick from — see the module docstring on why this never auto-links. """ probe = (needle or "").strip().lower() if len(probe) < 3: return [] probe = re.sub(r"^[a-z]+://", "", probe).split("/", 1)[0] rows = ( await db.execute( _base_query() .where( RegisteredSystem.status != const.STATUS_RETIRED, or_( func.lower( func.cast(RegisteredSystem.hostnames, Text), ).like(f"%{probe}%"), func.lower(RegisteredSystem.name).like(f"%{probe}%"), func.lower(RegisteredSystem.slug).like(f"%{probe}%"), ), ) .limit(limit), ) ).scalars().all() return list(rows) async def load_person_optional( db: AsyncSession, person_id: uuid.UUID, ) -> IdentityPerson | None: """The actor row, or ``None`` when the JWT outlived the person. Used by the router so a soft-deleted admin mid-session records a null actor instead of failing the write.""" return ( await db.execute( select(IdentityPerson) .where(IdentityPerson.id == person_id) .options(selectinload(IdentityPerson.org_memberships)), ) ).scalar_one_or_none() def _assert_assignable(person: IdentityPerson) -> IdentityPerson: """Refuse to name a departed employee as an owner. The bounty domain already refuses to ASSIGN a report to a non-active person; ownership is the stronger claim of the two and was letting a leaver through. A system owned by someone who has left is worse than an ownerless one: the ownerless read chases the blank, while a departed name satisfies every "has an owner" check and the offboarding blocker that would have caught it has long since completed. """ if person.employment_status != "active": raise ConflictError( f"{person.work_email} is {person.employment_status!r} — a system " "owner must be an active employee.", code="SYSTEM_OWNER_NOT_ACTIVE", ) return person async def load_person( db: AsyncSession, person_id: uuid.UUID, ) -> IdentityPerson: """Eager-load org memberships: ``IdentityPerson.department`` is a property over that relationship, so touching it on a lazily-loaded person raises ``MissingGreenlet`` under asyncio.""" person = ( await db.execute( select(IdentityPerson) .where( IdentityPerson.id == person_id, IdentityPerson.deleted_at.is_(None), ) .options(selectinload(IdentityPerson.org_memberships)), ) ).scalar_one_or_none() if person is None: raise NotFoundError("Person", str(person_id)) return person # ════════════════════════════════════════════════════════════════════ # Events # ════════════════════════════════════════════════════════════════════ async def add_event( db: AsyncSession, system: RegisteredSystem, *, event_type: str, actor: IdentityPerson | None = None, body: str | None = None, meta: dict[str, Any] | None = None, ) -> RegisteredSystemEvent: event = RegisteredSystemEvent( system_id=system.id, event_type=event_type, actor_person_id=actor.id if actor else None, actor_name=(actor.display_name if actor else None), body=body, meta=meta, ) db.add(event) return event async def list_events( db: AsyncSession, system_id: uuid.UUID, *, limit: int = 200, ) -> list[RegisteredSystemEvent]: rows = ( await db.execute( select(RegisteredSystemEvent) .where(RegisteredSystemEvent.system_id == system_id) # ``created_at`` is second-granular; the id tiebreak keeps # pagination stable when two events land in the same second. # The A→B chain itself is reconstructed from ``meta``, not # from row order. .order_by( RegisteredSystemEvent.created_at.desc(), RegisteredSystemEvent.id.desc(), ) .limit(limit), ) ).scalars().all() return list(rows) # ════════════════════════════════════════════════════════════════════ # Writes # ════════════════════════════════════════════════════════════════════ async def create_system( db: AsyncSession, *, name: str, slug: str | None = None, kind: str = const.KIND_OTHER, exposure: str = const.EXPOSURE_UNKNOWN, status: str = const.STATUS_ACTIVE, hostnames: list[str] | None = None, description: str | None = None, repo: str | None = None, owner_person_id: uuid.UUID | None = None, backup_owner_person_id: uuid.UUID | None = None, budget_track: str = const.TRACK_PRODUCT, in_bounty_scope: bool = False, notes: str | None = None, actor: IdentityPerson | None = None, ) -> RegisteredSystem: clean_name = _bounded(name, const.MAX_NAME_CHARS, "name") if not clean_name: raise ValidationError("name is required") final_slug = normalize_slug(slug or clean_name) if await get_by_slug(db, final_slug) is not None: raise ConflictError( f"a system with slug '{final_slug}' already exists", code="SYSTEM_SLUG_TAKEN", ) owner = ( _assert_assignable(await load_person(db, owner_person_id)) if owner_person_id else None ) backup = ( _assert_assignable(await load_person(db, backup_owner_person_id)) if backup_owner_person_id else None ) if owner and backup and owner.id == backup.id: raise ValidationError( "backup owner must be a different person than the owner", ) # No owner given? Ask DNS. A system registered from a hostname that # turned up in a researcher's mail has nobody typing a form, and the # register's failure mode is blanks — six of the first eight had one. # Only an unambiguous STRONG record qualifies (see the coordinator): # a zone-level record's owner administers the ZONE, and auto-filling # those would invent owners. Recorded as its own event so the trail # says where the name came from and one click undoes it. auto_owner: dict[str, Any] | None = None if owner is None: from app.core.coordinators import ( # noqa: PLC0415 dns_owner_coordinator as dns_owner, ) auto_owner = await dns_owner.resolve_auto_owner( db, list(normalize_hostnames(hostnames) or []), ) if auto_owner is not None: try: owner = _assert_assignable( await load_person(db, uuid.UUID(auto_owner["person_id"])), ) except (ConflictError, NotFoundError, ValueError): # The DNS record points at somebody Helix will not accept # as an owner (departed between the two reads, or an id # that no longer resolves). Stay ownerless and visible # rather than half-assigned. owner, auto_owner = None, None system = RegisteredSystem( slug=final_slug, name=clean_name, kind=_validate_enum(kind, const.KINDS, "kind"), exposure=_validate_enum(exposure, const.EXPOSURES, "exposure"), status=_validate_enum(status, const.STATUSES, "status"), hostnames=normalize_hostnames(hostnames) or None, description=_bounded(description, const.MAX_NOTES_CHARS, "description"), repo=_bounded(repo, 255, "repo"), owner_person_id=owner.id if owner else None, backup_owner_person_id=backup.id if backup else None, owner_department=(owner.department if owner else None), owner_assigned_at=_naive(_now()) if owner else None, budget_track=_validate_enum(budget_track, const.TRACKS, "budget_track"), in_bounty_scope=bool(in_bounty_scope), notes=_bounded(notes, const.MAX_NOTES_CHARS, "notes"), source="manual", created_by=actor.id if actor else None, ) db.add(system) await db.flush() await add_event( db, system, event_type=const.EVENT_CREATED, actor=actor, meta={"slug": final_slug, "owner_person_id": str(owner.id) if owner else None}, ) if auto_owner is not None: # Distinct from the human path: "DNS said so" is a weaker claim # than "a person decided", and the reader of an ownership dispute # needs to be able to tell them apart. await add_event( db, system, event_type=const.EVENT_OWNER_ASSIGNED, actor=actor, meta={ "from_person_id": None, "from_name": None, "to_person_id": auto_owner["person_id"], "to_name": auto_owner["display_name"], "source": "dns_record", "dns_hostname": auto_owner["hostname"], "dns_record_types": auto_owner["record_types"], }, ) elif owner is not None: # The ownership trail must be readable on its own: filtering the # events to ``owner_assigned`` has to answer "who has been # responsible, since when" without the reader also knowing that # ``created.meta`` happens to carry an owner. One event kind, one # question. await add_event( db, system, event_type=const.EVENT_OWNER_ASSIGNED, actor=actor, meta={ "from_person_id": None, "from_name": None, "to_person_id": str(owner.id), "to_name": owner.display_name, }, ) await db.commit() return await get_system(db, system.id) _MUTABLE_FIELDS = ( "name", "kind", "exposure", "status", "hostnames", "description", "repo", "budget_track", "in_bounty_scope", "notes", ) async def update_system( db: AsyncSession, system_id: uuid.UUID, *, payload: dict[str, Any], actor: IdentityPerson | None = None, ) -> RegisteredSystem: """Patch the descriptive fields. Ownership is deliberately NOT settable here — it goes through :func:`transfer_owner` so that every ownership change is guaranteed to leave a trail.""" system = await get_system(db, system_id) changes: dict[str, Any] = {} for field in _MUTABLE_FIELDS: if field not in payload: continue value = payload[field] if field == "name": value = _bounded(value, const.MAX_NAME_CHARS, "name") if not value: raise ValidationError("name cannot be blank") elif field == "kind": value = _validate_enum(value, const.KINDS, "kind") elif field == "exposure": value = _validate_enum(value, const.EXPOSURES, "exposure") elif field == "status": value = _validate_enum(value, const.STATUSES, "status") elif field == "budget_track": value = _validate_enum(value, const.TRACKS, "budget_track") elif field == "hostnames": value = normalize_hostnames(value) or None elif field == "in_bounty_scope": value = bool(value) elif field == "repo": value = _bounded(value, 255, "repo") else: value = _bounded(value, const.MAX_NOTES_CHARS, field) current = getattr(system, field) if current != value: changes[field] = {"from": current, "to": value} setattr(system, field, value) if not changes: return system if "status" in changes: await add_event( db, system, event_type=const.EVENT_STATUS_CHANGED, actor=actor, meta=changes["status"], ) descriptive = {k: v for k, v in changes.items() if k != "status"} if descriptive: await add_event( db, system, event_type=const.EVENT_UPDATED, actor=actor, meta={"fields": sorted(descriptive)}, ) await db.commit() return await get_system(db, system_id) async def transfer_owner( db: AsyncSession, system_id: uuid.UUID, *, owner_person_id: uuid.UUID | None, backup_owner_person_id: uuid.UUID | None = None, set_backup: bool = False, note: str | None = None, actor: IdentityPerson | None = None, ) -> RegisteredSystem: """Set or clear the owner (and optionally the backup owner). ``set_backup`` distinguishes "leave the backup alone" from "clear the backup" — one nullable argument cannot express both, and defaulting a clear would wipe the successor the offboarding flow just recorded. """ system = await get_system(db, system_id) previous_id = system.owner_person_id previous_name = system.owner.display_name if system.owner else None new_owner = ( _assert_assignable(await load_person(db, owner_person_id)) if owner_person_id else None ) new_backup = system.backup_owner if set_backup: new_backup = ( _assert_assignable(await load_person(db, backup_owner_person_id)) if backup_owner_person_id else None ) if new_owner and new_backup and new_owner.id == new_backup.id: # Promoting the backup to owner is the common case: clear the # backup slot rather than rejecting the transfer, otherwise the # offboarding handover needs two round-trips to succeed. new_backup = None set_backup = True if new_owner is not None and previous_id == new_owner.id and not set_backup: return system system.owner_person_id = new_owner.id if new_owner else None system.owner_department = new_owner.department if new_owner else None system.owner_assigned_at = _naive(_now()) if new_owner else None if set_backup: system.backup_owner_person_id = new_backup.id if new_backup else None # An owner-unchanged call (the backup slot was the only thing touched) # must NOT write an ownership event: a timeline row reading "owner # changed: X → X" is a false statement about the responsibility # chain, and the trail is read as fact during an incident. owner_unchanged = previous_id == (new_owner.id if new_owner else None) if owner_unchanged: event_type = const.EVENT_UPDATED meta: dict[str, Any] = { "fields": ["backup_owner"], "backup_owner_person_id": ( str(new_backup.id) if new_backup else None ), } else: event_type = ( const.EVENT_OWNER_CLEARED if new_owner is None else const.EVENT_OWNER_ASSIGNED ) meta = { "from_person_id": str(previous_id) if previous_id else None, "from_name": previous_name, "to_person_id": str(new_owner.id) if new_owner else None, "to_name": new_owner.display_name if new_owner else None, } await add_event( db, system, event_type=event_type, actor=actor, body=_bounded(note, const.MAX_NOTES_CHARS, "note"), meta=meta, ) await db.commit() return await get_system(db, system_id) async def retire_system( db: AsyncSession, system_id: uuid.UUID, *, actor: IdentityPerson | None = None, note: str | None = None, ) -> RegisteredSystem: """Mark the system retired. Status only — the row is NOT soft-deleted. Retiring is a lifecycle statement, not a mistake being erased. Keeping the row readable means a historical bounty report still resolves to a name, the ownership trail stays answerable, and an operator who retired the wrong thing can set the status back. ``deleted_at`` is reserved for a genuine purge. Every governance read excludes it through ``LIVE_STATUSES`` rather than through the soft-delete filter, so a retired system stops counting as surface and stops demanding an owner without disappearing. """ system = await get_system(db, system_id) if system.status == const.STATUS_RETIRED: return system previous = system.status system.status = const.STATUS_RETIRED await add_event( db, system, event_type=const.EVENT_RETIRED, actor=actor, body=note, meta={"from": previous, "to": const.STATUS_RETIRED}, ) await db.commit() return await get_system(db, system_id) # ════════════════════════════════════════════════════════════════════ # Cross-domain reads (consumed by offboarding + bounty via coordinators) # ════════════════════════════════════════════════════════════════════ async def systems_owned_by( db: AsyncSession, person_id: uuid.UUID, *, include_backup: bool = True, ) -> list[RegisteredSystem]: """Live systems this person is accountable for. Retired systems are excluded: forcing a handover of something that no longer runs turns the offboarding blocker into noise, and noise is how a real blocker gets clicked past. """ conditions = [RegisteredSystem.owner_person_id == person_id] if include_backup: conditions.append( RegisteredSystem.backup_owner_person_id == person_id, ) rows = ( await db.execute( _base_query() .where( RegisteredSystem.status.in_(const.LIVE_STATUSES), or_(*conditions), ) .order_by(RegisteredSystem.name.asc()), ) ).scalars().all() return list(rows) async def coverage_summary(db: AsyncSession) -> dict[str, Any]: """Counts the registry admin page leads with. The denominator is live systems only — a retired system without an owner is not a governance gap, and counting it would make the number drift down every time something is switched off correctly. """ # Outer-joined to the owner so "is the named owner still here" is one # query rather than one per row — and so a NULL owner and a departed # owner stay distinguishable in the same pass. owner = aliased(IdentityPerson) rows = ( await db.execute( select( RegisteredSystem.status, RegisteredSystem.exposure, RegisteredSystem.owner_person_id, RegisteredSystem.backup_owner_person_id, RegisteredSystem.in_bounty_scope, ( (owner.employment_status == "active") & owner.deleted_at.is_(None) ).label("owner_active"), ) .outerjoin(owner, owner.id == RegisteredSystem.owner_person_id) .where(RegisteredSystem.deleted_at.is_(None)), ) ).all() live = [r for r in rows if r.status in const.LIVE_STATUSES] # "Owned" means an ACTIVE person answers for it. Counting a departed # owner as covered is how a 100% coverage number coexists with nobody # being accountable — the row satisfies every ``owner IS NOT NULL`` # check while the human behind it left months ago. They are reported # as their own bucket rather than folded into ``ownerless``, because # the fix differs: an ownerless system needs a first owner, a departed # one needs a handover somebody forgot to do. owned = [ r for r in live if r.owner_person_id is not None and r.owner_active ] departed = [ r for r in live if r.owner_person_id is not None and not r.owner_active ] # ``total`` counts every registered row including retired ones (they # are status-retired, not soft-deleted), while every governance ratio # below divides by ``live``. return { "total": len(rows), "live": len(live), "owned": len(owned), "ownerless": len([r for r in live if r.owner_person_id is None]), "owner_departed": len(departed), "coverage_pct": ( round(len(owned) * 100 / len(live), 1) if live else 0.0 ), "with_backup": len( [r for r in live if r.backup_owner_person_id is not None], ), "external": len( [r for r in live if r.exposure in const.EXTERNAL_EXPOSURES], ), # An externally reachable system whose owner has left is the same # operational gap as one that never had an owner: there is nobody # to route a finding to. Both count here. "external_ownerless": len( [ r for r in live if r.exposure in const.EXTERNAL_EXPOSURES and (r.owner_person_id is None or not r.owner_active) ], ), "in_bounty_scope": len([r for r in live if r.in_bounty_scope]), "retired": len( [r for r in rows if r.status == const.STATUS_RETIRED], ), }