"""HTTP API for the relationship graph. Endpoints: * ``GET /api/v1/relationships/graph?center_type=...¢er_id=...&depth=...`` BFS-extracted sub-graph centred on a node. Used by the ``/admin/relationship-graph`` page (admin/auditor centred at arbitrary node) and the ``/me`` embed (employee centred at self). * ``GET /api/v1/relationships/sync-status`` Read-only freshness signal — when did each of the two scheduled sync jobs last run, and what's the next interval-driven tick? Used by RelationshipGraphPage to render the "last synced N min ago" chip so users don't have to wonder whether a fresh mutation has landed in the graph yet. Available to anyone who can view the graph (admin / auditor / cloud_admin / asset_admin / hr / finance / and self-only employees), since it doesn't disclose any data — only timestamps about background jobs. * ``POST /api/v1/relationships/sync-now`` (admin) Force a full sync immediately. Kept for genuine admin recovery scenarios (e.g. after a backup-restore that left the materialised cache stale by more than the 5-min incremental window). The employee-facing graph page no longer surfaces this — incremental cadence is 5 min so the manual button isn't part of normal flow. """ from __future__ import annotations import uuid from datetime import UTC, datetime from typing import Any from fastapi import APIRouter, Depends, Query, Request from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.database import get_db from app.core.dependencies import get_current_user from app.core.exceptions import ForbiddenError, ValidationError from app.core.rbac import ( GLOBAL_ADMIN_ROLE, can_view_relationship_graph, has_role, ) from app.core.response import success_response from app.domains.relationships.constants import ( ALL_EDGE_TYPES, ALL_NODE_TYPES, NODE_TYPE_API_KEY, NODE_TYPE_CLOUD_GRANT, NODE_TYPE_CLOUD_SCOPE, NODE_TYPE_CLOUD_SERVER, NODE_TYPE_DNS_RECORD, NODE_TYPE_DOMAIN, NODE_TYPE_EMAIL_GROUP, NODE_TYPE_HARDWARE_ASSET, NODE_TYPE_PERMISSION, NODE_TYPE_PERSON, NODE_TYPE_SOFTWARE, NODE_TYPE_SOFTWARE_SEAT, REL_ASSIGNED_TO, REL_GRANTS_ACCESS_TO, REL_HAS_API_KEY, REL_HOLDS_SEAT, REL_HOSTS, REL_OWNS, REL_RESOLVES_DNS, ) from app.domains.relationships.query_service import ( DEFAULT_MAX_NODES, DEFAULT_PER_NODE_FANOUT_CAP, RelationshipQueryService, ) from app.domains.relationships.sync_service import RelationshipSyncService from app.identity.person_models import IdentityPerson from app.scheduled_jobs.models import ScheduledJob router = APIRouter( prefix="/api/v1/relationships", tags=["relationships"], ) # Edge types a non-privileged employee may traverse on the /me embed. # Excludes management-chain edges (``REL_MANAGES`` / ``REL_REPORTS_TO``) # and department-membership edges (``REL_MEMBER_OF`` / ``REL_LEADS``) # which would let an employee BFS their own manager → every peer → # their teams and scrape display_name + work_email of the entire # org. The platform deliberately keeps ``GET /api/v1/users`` (full # directory) admin-only; the graph endpoint must not be a # back-door around that. # # Also excludes ``REL_RESPONSIBLE_FOR`` / ``REL_IT_CONTACT_FOR`` — # those are software-catalog ownership edges centred on a person, # but a self-only employee shouldn't enumerate "every IT contact's # label" via the catalog. They can still see their own assigned # resources (assets / seats / api_keys / cloud scopes / domains # they own) one hop out. _SELF_GRAPH_SAFE_REL_TYPES = frozenset({ REL_ASSIGNED_TO, REL_HOLDS_SEAT, REL_HAS_API_KEY, REL_OWNS, REL_GRANTS_ACCESS_TO, REL_HOSTS, REL_RESOLVES_DNS, }) # Hard depth ceiling for the self-only path — even at depth=1 the # centre node's direct neighbours are the user's own assets / seats / # scopes, which is the legitimate /me-embed use case. depth>=2 starts # walking into resource→other-person edges (e.g. an asset previously # assigned to a peer) and re-introduces the leak vector. _SELF_GRAPH_MAX_DEPTH = 1 # Node types a non-privileged employee may surface on the /me embed. # Pairs with ``_SELF_GRAPH_SAFE_REL_TYPES`` for defense-in-depth: the # rel-type filter is the primary gate (rel_types in this set don't # emit person→person edges today), but the node-type whitelist below # guarantees that even a future rel-type schema change (e.g. a new # "delegate-of" person↔person edge added to ``REL_OWNS``) cannot # leak ``NODE_TYPE_PERSON`` neighbours through this path. Person # neighbours stay reachable through the centre node itself (always # included as the focused person) but never as BFS-expanded # directory entries. ``NODE_TYPE_DEPARTMENT`` is also excluded so # the org structure cannot be reconstructed from the graph either. _SELF_GRAPH_SAFE_NODE_TYPES = frozenset({ NODE_TYPE_PERSON, # the centre node itself NODE_TYPE_HARDWARE_ASSET, NODE_TYPE_SOFTWARE, NODE_TYPE_SOFTWARE_SEAT, NODE_TYPE_API_KEY, NODE_TYPE_PERMISSION, NODE_TYPE_EMAIL_GROUP, NODE_TYPE_CLOUD_SERVER, NODE_TYPE_CLOUD_SCOPE, NODE_TYPE_CLOUD_GRANT, NODE_TYPE_DOMAIN, NODE_TYPE_DNS_RECORD, }) _FULL_SYNC_NAME = "relationships.full_sync" _INCREMENTAL_SYNC_NAME = "relationships.incremental_sync" def _serialize_sync_job(job: ScheduledJob | None) -> dict[str, Any] | None: """Project the small slice of ``ScheduledJob`` the UI needs. Intentionally narrow — we expose timestamps + cadence + pass/fail state, never the full ``last_result_summary`` (which can leak internal counts), the description, or the config blob. """ if job is None: return None return { "name": job.name, "enabled": job.enabled, "interval_seconds": job.interval_seconds, "cron_expr": job.cron_expr, "last_run_at": job.last_run_at.isoformat() if job.last_run_at else None, "last_status": job.last_status, } @router.get("/graph") async def get_graph( request: Request, center_type: str = Query(...), center_id: str = Query(...), depth: int = Query(default=2, ge=0, le=5), rel_types: str | None = Query(default=None), node_types: str | None = Query(default=None), fanout: int = Query( default=DEFAULT_PER_NODE_FANOUT_CAP, ge=1, le=200, ), max_nodes: int = Query(default=DEFAULT_MAX_NODES, ge=1, le=2000), user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Return a sub-graph centred on the given node. Authorisation: * Admin / auditor / finance / asset_admin / cloud_admin / hr — any node permitted (``can_view_relationship_graph`` covers this). * Regular employees — only allowed when the centre is themselves (``center_type=person, center_id=``). The /me embed uses this path. """ is_privileged = can_view_relationship_graph(user.role_names) if is_privileged: pass elif center_type == "person" and str(user.id) == center_id: pass else: raise ForbiddenError( "Cannot view relationship graph centred on someone else" ) rt_list = ( [s.strip() for s in rel_types.split(",") if s.strip()] if rel_types else None ) nt_list = ( [s.strip() for s in node_types.split(",") if s.strip()] if node_types else None ) # Non-privileged self-only callers: clamp depth + restrict edge # filter to the "my own resources" set. Without this guard an # employee can ``depth=5, fanout=200, max_nodes=2000`` from their # own person node, follow ``REL_MANAGES`` / ``REL_REPORTS_TO`` / # ``REL_MEMBER_OF`` up the chain, and scrape display_name + # work_email of every peer reachable in five hops — a textbook # IDOR-via-graph-traversal pattern that re-leaks the directory # ``GET /api/v1/users`` is admin-only specifically to protect. # Pre-launch hardening 2026-05-09 (round 16 audit). if not is_privileged: depth = min(depth, _SELF_GRAPH_MAX_DEPTH) if rt_list is None: rt_list = sorted(_SELF_GRAPH_SAFE_REL_TYPES) else: rt_list = [ rt for rt in rt_list if rt in _SELF_GRAPH_SAFE_REL_TYPES ] # Reject when the caller explicitly asked for ONLY org- # chart edges. Returning an empty list silently would mask # the gate rather than surface "you can't see that here". if not rt_list: raise ForbiddenError( "Self-only graph view cannot include " "management-chain or department-membership edges. " "Ask an admin if you need org-structure visibility." ) # Belt-and-suspenders: also clamp ``node_types`` so a future # schema change that turns a currently-resource-only rel into # a person↔person edge can't silently re-leak directory data # via this endpoint. The person neighbours that exist today # (manager / subordinate / peer) all sit on rel_types we # already filter out above, so this is only a fragility- # prevention guard. Same 403-on-empty stance as rel_types. if nt_list is None: nt_list = sorted(_SELF_GRAPH_SAFE_NODE_TYPES) else: nt_list = [ nt for nt in nt_list if nt in _SELF_GRAPH_SAFE_NODE_TYPES ] if not nt_list: raise ForbiddenError( "Self-only graph view cannot project onto " "directory node types. Ask an admin if you need " "person- or department-level visibility." ) # Reject unknown rel_types / node_types up front so a typo or a # client passing junk surfaces as 422 rather than silently returning # an empty graph (which used to look indistinguishable from "no # data" in the UI). if rt_list: unknown_rt = [r for r in rt_list if r not in ALL_EDGE_TYPES] if unknown_rt: raise ValidationError( f"Unknown rel_types: {', '.join(sorted(unknown_rt))}", ) if nt_list: unknown_nt = [n for n in nt_list if n not in ALL_NODE_TYPES] if unknown_nt: raise ValidationError( f"Unknown node_types: {', '.join(sorted(unknown_nt))}", ) if center_type not in ALL_NODE_TYPES: raise ValidationError(f"Unknown center_type: {center_type!r}") service = RelationshipQueryService(db) sub = await service.get_subgraph( center_type=center_type, center_id=center_id, depth=depth, rel_types=rt_list, node_types=nt_list, per_node_fanout_cap=fanout, max_nodes=max_nodes, ) return success_response( data={ "center": {"type": sub.center[0], "id": sub.center[1]}, "depth": sub.depth, "nodes": [ { "type": n.type, "id": n.id, "label": n.label, "attributes": n.attributes, } for n in sub.nodes ], "edges": [ { "from": {"type": e.from_type, "id": e.from_id}, "to": {"type": e.to_type, "id": e.to_id}, "rel_type": e.rel_type, "weight": e.weight, "attributes": e.attributes, } for e in sub.edges ], "truncated": sub.truncated, }, request=request, ) @router.get("/sync-status") async def get_sync_status( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Freshness signal for the relationship graph. Returns the last-run / next-run state of the two scheduled sync jobs plus the server's current UTC clock so the client can compute "synced N seconds ago" without trusting client-side time. Open to any authenticated user (no extra RBAC check) because: * The response carries no business data — only ``last_run_at`` and ``last_status`` for two named scheduled jobs. * Self-only employees view the graph via the ``/me`` embed and need this endpoint to render the freshness chip; gating on ``can_view_relationship_graph`` would force them into a falsely-blank chip and require us to hard-code "5 min" on the client. Better to share the truth. The ``user`` dep is kept (not removed) because ``get_current_user`` enforces "must be authenticated" — anonymous callers still 401 here, just like every other endpoint. We don't reference ``user`` in the body because RBAC isn't applied (see above). """ _ = user # Reference to satisfy unused-arg linters. rows = ( await db.execute( select(ScheduledJob).where( ScheduledJob.name.in_( [_FULL_SYNC_NAME, _INCREMENTAL_SYNC_NAME], ), ) ) ).scalars().all() by_name = {job.name: job for job in rows} return success_response( data={ "server_time": datetime.now(UTC).isoformat(), "full_sync": _serialize_sync_job(by_name.get(_FULL_SYNC_NAME)), "incremental_sync": _serialize_sync_job( by_name.get(_INCREMENTAL_SYNC_NAME), ), }, request=request, ) @router.post("/sync-now") async def sync_now( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): if not has_role(user.role_names, GLOBAL_ADMIN_ROLE): raise ForbiddenError( "Only global_admin can trigger an immediate full graph sync" ) summary = await RelationshipSyncService(db).run_full_sync() # A lock-busy run returns a ``skipped`` summary (no exception). Logging # "success" then would claim a manual sync that never ran — reflect the # real outcome so the audit trail is honest. skipped = bool(summary.get("skipped")) audit = AuditService(db) await audit.log( trace_id=getattr(request.state, "trace_id", None) or str(uuid.uuid4()), actor_type="user", actor_user_id=user.id, action="relationships.sync_now", resource_type="rel_relationships", resource_id="full_sync", status="skipped" if skipped else "success", detail={"summary_totals": summary.get("totals", {}), "skipped": skipped}, ) return success_response(data=summary, request=request)