|
|
@@ -0,0 +1,570 @@
|
|
|
+"""Repository contracts and a deterministic test adapter for enterprise identity."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import copy
|
|
|
+import json
|
|
|
+import secrets
|
|
|
+import threading
|
|
|
+from collections import defaultdict
|
|
|
+from collections.abc import Callable
|
|
|
+from typing import Any
|
|
|
+
|
|
|
+from sqlalchemy import text
|
|
|
+
|
|
|
+from app.core.common.identifiers import new_governance_uid
|
|
|
+from app.core.system.enterprise_identity import IdentityPolicyError
|
|
|
+
|
|
|
+
|
|
|
+class MemoryIdentityRepository:
|
|
|
+ """Small adapter used by protocol tests; production persistence is PostgreSQL."""
|
|
|
+
|
|
|
+ def __init__(self) -> None:
|
|
|
+ self.flows: dict[str, dict[str, Any]] = {}
|
|
|
+ self.sessions: dict[str, dict[str, Any]] = {}
|
|
|
+ self.directory_events: dict[tuple[str, str, str], dict[str, Any]] = {}
|
|
|
+ self.directory_checkpoints: dict[tuple[str, str], dict[str, Any]] = {}
|
|
|
+ self.directory_conflicts: list[dict[str, Any]] = []
|
|
|
+ self.identities: dict[tuple[str, str], dict[str, Any]] = {}
|
|
|
+ self.organization_nodes: dict[tuple[str, str, str], dict[str, Any]] = {}
|
|
|
+ self.emergency_requests: dict[str, dict[str, Any]] = {}
|
|
|
+ self.audits: list[dict[str, Any]] = []
|
|
|
+ self.refresh_index: dict[str, str] = {}
|
|
|
+ self.family_members: dict[str, set[str]] = defaultdict(set)
|
|
|
+ self._session_lock = threading.RLock()
|
|
|
+ self._directory_lock = threading.RLock()
|
|
|
+
|
|
|
+ def put_flow(self, flow: dict[str, Any]) -> None:
|
|
|
+ self.flows[flow["state_hash"]] = copy.deepcopy(flow)
|
|
|
+
|
|
|
+ def consume_flow(self, state_hash: str, *, commit: bool = True) -> dict[str, Any] | None:
|
|
|
+ del commit
|
|
|
+ flow = self.flows.get(state_hash)
|
|
|
+ if not flow or flow.get("consumed_at"):
|
|
|
+ return None
|
|
|
+ flow["consumed_at"] = True
|
|
|
+ return copy.deepcopy(flow)
|
|
|
+
|
|
|
+ def put_session(self, session: dict[str, Any]) -> None:
|
|
|
+ self.sessions[session["uid"]] = copy.deepcopy(session)
|
|
|
+ self.refresh_index[session["refresh_hash"]] = session["uid"]
|
|
|
+ self.family_members[session["family_uid"]].add(session["uid"])
|
|
|
+
|
|
|
+ def create_session(self, session: dict[str, Any], *, max_sessions: int, commit: bool = True) -> None:
|
|
|
+ del commit
|
|
|
+ with self._session_lock:
|
|
|
+ self.put_session(session)
|
|
|
+ active = sorted((x for x in self.sessions.values()
|
|
|
+ if x["user_uid"] == session["user_uid"] and x["status"] == "active"),
|
|
|
+ key=lambda x: (x["created_at"], x["uid"]), reverse=True)
|
|
|
+ for old in active[max_sessions:]:
|
|
|
+ self.update_session(old["uid"], status="revoked", revoke_reason="concurrency_limit")
|
|
|
+
|
|
|
+ def rotate_refresh(self, old_hash: str, new_session: dict[str, Any]) -> str:
|
|
|
+ old = self.get_session_by_refresh(old_hash)
|
|
|
+ if not old:
|
|
|
+ return "unknown"
|
|
|
+ if old["status"] != "active":
|
|
|
+ self.revoke_family(old["family_uid"], reason="refresh_reuse")
|
|
|
+ return "reuse"
|
|
|
+ self.update_session(old["uid"], status="rotated", rotated_at=new_session["created_at"])
|
|
|
+ self.put_session(new_session)
|
|
|
+ return "rotated"
|
|
|
+
|
|
|
+ def get_session(self, uid: str) -> dict[str, Any] | None:
|
|
|
+ value = self.sessions.get(uid)
|
|
|
+ return copy.deepcopy(value) if value else None
|
|
|
+
|
|
|
+ def get_session_by_refresh(self, refresh_hash: str) -> dict[str, Any] | None:
|
|
|
+ uid = self.refresh_index.get(refresh_hash)
|
|
|
+ return self.get_session(uid) if uid else None
|
|
|
+
|
|
|
+ def update_session(self, uid: str, **changes: Any) -> dict[str, Any]:
|
|
|
+ self.sessions[uid].update(copy.deepcopy(changes))
|
|
|
+ return self.get_session(uid) or {}
|
|
|
+
|
|
|
+ def list_sessions(self, provider_uid: str | None = None, subject: str | None = None) -> list[dict[str, Any]]:
|
|
|
+ rows = self.sessions.values()
|
|
|
+ if provider_uid is not None:
|
|
|
+ rows = (item for item in rows if item.get("provider_uid") == provider_uid)
|
|
|
+ if subject is not None:
|
|
|
+ rows = (item for item in rows if item["subject"] == subject)
|
|
|
+ return [copy.deepcopy(item) for item in rows]
|
|
|
+
|
|
|
+ def revoke_family(self, family_uid: str, *, reason: str) -> None:
|
|
|
+ for uid in self.family_members.get(family_uid, ()):
|
|
|
+ self.sessions[uid].update(status="revoked", revoke_reason=reason)
|
|
|
+
|
|
|
+ def revoke_subject_sessions(self, provider_uid: str | None, subject: str, *, reason: str,
|
|
|
+ commit: bool = True) -> None:
|
|
|
+ del commit
|
|
|
+ for record in self.list_sessions(provider_uid, subject):
|
|
|
+ if record["status"] == "active":
|
|
|
+ self.update_session(record["uid"], status="revoked", revoke_reason=reason)
|
|
|
+
|
|
|
+ def get_directory_event(self, provider_uid: str, source: str, source_event_id: str) -> dict[str, Any] | None:
|
|
|
+ value = self.directory_events.get((provider_uid, source, source_event_id))
|
|
|
+ return copy.deepcopy(value) if value else None
|
|
|
+
|
|
|
+ def put_directory_event(self, event: dict[str, Any]) -> None:
|
|
|
+ self.directory_events[(event["provider_uid"], event["source"], event["source_event_id"])] = copy.deepcopy(event)
|
|
|
+ self.directory_checkpoints[(event["provider_uid"], event["source"])] = {
|
|
|
+ "cursor": event["cursor"], "cursor_sequence": event["cursor_sequence"]}
|
|
|
+
|
|
|
+ def get_directory_checkpoint(self, provider_uid: str, source: str) -> dict[str, Any] | None:
|
|
|
+ value = self.directory_checkpoints.get((provider_uid, source))
|
|
|
+ return copy.deepcopy(value) if value else None
|
|
|
+
|
|
|
+ def record_directory_conflict(self, **record: Any) -> None:
|
|
|
+ self.directory_conflicts.append(copy.deepcopy(record))
|
|
|
+
|
|
|
+ def put_identity(self, provider_uid: str, subject: str, identity: dict[str, Any], *, commit: bool = True) -> None:
|
|
|
+ del commit
|
|
|
+ self.identities[(provider_uid, subject)] = copy.deepcopy(identity)
|
|
|
+
|
|
|
+ def get_identity(self, provider_uid: str, subject: str, *, for_update: bool = False) -> dict[str, Any] | None:
|
|
|
+ del for_update
|
|
|
+ value = self.identities.get((provider_uid, subject))
|
|
|
+ return copy.deepcopy(value) if value else None
|
|
|
+
|
|
|
+ def get_organization_node(self, provider_uid: str, node_type: str, external_id: str,
|
|
|
+ *, for_update: bool = False) -> dict[str, Any] | None:
|
|
|
+ del for_update
|
|
|
+ value = self.organization_nodes.get((provider_uid, node_type, external_id))
|
|
|
+ return copy.deepcopy(value) if value else None
|
|
|
+
|
|
|
+ def put_organization_node(self, node: dict[str, Any], *, commit: bool = True) -> None:
|
|
|
+ del commit
|
|
|
+ key = (node["provider_uid"], node["node_type"], node["external_id"])
|
|
|
+ self.organization_nodes[key] = copy.deepcopy(node)
|
|
|
+
|
|
|
+ def apply_directory_event_atomic(self, event: dict[str, Any],
|
|
|
+ mutation: Callable[[], dict[str, Any]]) -> dict[str, Any]:
|
|
|
+ """Apply the lifecycle mutation and checkpoint as one in-memory unit of work."""
|
|
|
+ with self._directory_lock:
|
|
|
+ previous = self.get_directory_event(event["provider_uid"], event["source"], event["source_event_id"])
|
|
|
+ if previous:
|
|
|
+ if previous["payload_digest"] != event["payload_digest"]:
|
|
|
+ self.record_directory_conflict(**event, reason="source_event_id_payload_mismatch")
|
|
|
+ raise IdentityPolicyError("directory event idempotency conflict")
|
|
|
+ return {**previous["result"], "idempotent": True}
|
|
|
+ checkpoint = self.get_directory_checkpoint(event["provider_uid"], event["source"])
|
|
|
+ if checkpoint and event["cursor_sequence"] <= checkpoint["cursor_sequence"]:
|
|
|
+ self.record_directory_conflict(**event, reason="cursor_not_monotonic")
|
|
|
+ raise IdentityPolicyError("directory cursor is not monotonic")
|
|
|
+ snapshot = copy.deepcopy((self.identities, self.organization_nodes, self.sessions,
|
|
|
+ self.refresh_index, self.family_members,
|
|
|
+ self.directory_events, self.directory_checkpoints))
|
|
|
+ try:
|
|
|
+ result = mutation()
|
|
|
+ self.put_directory_event({**event, "result": result})
|
|
|
+ return result
|
|
|
+ except Exception:
|
|
|
+ (self.identities, self.organization_nodes, self.sessions, self.refresh_index,
|
|
|
+ self.family_members, self.directory_events, self.directory_checkpoints) = snapshot
|
|
|
+ raise
|
|
|
+
|
|
|
+ def put_emergency(self, request: dict[str, Any]) -> None:
|
|
|
+ self.emergency_requests[request["uid"]] = copy.deepcopy(request)
|
|
|
+ if request["status"] in {"closed", "expired", "reviewed"}:
|
|
|
+ for session in self.sessions.values():
|
|
|
+ if session.get("emergency_request_uid") == request["uid"] and session["status"] == "active":
|
|
|
+ session.update(status="revoked", revoke_reason=f"emergency_{request['status']}")
|
|
|
+
|
|
|
+ def get_emergency(self, uid: str) -> dict[str, Any] | None:
|
|
|
+ value = self.emergency_requests.get(uid)
|
|
|
+ return copy.deepcopy(value) if value else None
|
|
|
+
|
|
|
+ def audit(self, event: dict[str, Any]) -> None:
|
|
|
+ self.audits.append(copy.deepcopy(event))
|
|
|
+
|
|
|
+ def emergency_session_active(self, request_uid: str, account_uid: str, now: Any) -> bool:
|
|
|
+ request = self.emergency_requests.get(request_uid)
|
|
|
+ return bool(request and request["account_uid"] == account_uid and request["status"] == "active"
|
|
|
+ and request["requested_at"] <= now < request["expires_at"])
|
|
|
+
|
|
|
+
|
|
|
+class PostgresIdentityRepository:
|
|
|
+ """PostgreSQL adapter for OIDC flows and server-side sessions."""
|
|
|
+
|
|
|
+ def __init__(self, session: Any) -> None:
|
|
|
+ self.session = session
|
|
|
+
|
|
|
+ def put_flow(self, flow: dict[str, Any]) -> None:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_oidc_flows
|
|
|
+ (uid,state_hash,provider_uid,provider_version,nonce_hash,pkce_verifier_hash,redirect_uri,status,expires_at)
|
|
|
+ VALUES (CAST(:uid AS uuid),:state_hash,CAST(:provider_uid AS uuid),:provider_version,:nonce_hash,
|
|
|
+ :verifier_hash,:redirect_uri,'pending',:expires_at)
|
|
|
+ """), {**flow, "uid": new_governance_uid(), "verifier_hash": flow["verifier_hash"]})
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def consume_flow(self, state_hash: str, *, commit: bool = True) -> dict[str, Any] | None:
|
|
|
+ row = self.session.execute(text("""
|
|
|
+ UPDATE public.identity_oidc_flows SET status='consumed', consumed_at=CURRENT_TIMESTAMP
|
|
|
+ WHERE state_hash=:state_hash AND status='pending' AND expires_at>CURRENT_TIMESTAMP
|
|
|
+ RETURNING provider_uid::text,provider_version,nonce_hash,pkce_verifier_hash,redirect_uri,expires_at
|
|
|
+ """), {"state_hash": state_hash}).mappings().one_or_none()
|
|
|
+ if commit:
|
|
|
+ self.session.commit()
|
|
|
+ return ({"provider_uid": row["provider_uid"], "provider_version": row["provider_version"],
|
|
|
+ "nonce_hash": row["nonce_hash"], "verifier_hash": row["pkce_verifier_hash"],
|
|
|
+ "redirect_uri": row["redirect_uri"], "expires_at": row["expires_at"]} if row else None)
|
|
|
+
|
|
|
+ def _insert_session(self, record: dict[str, Any]) -> None:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_sessions
|
|
|
+ (uid,family_uid,identity_link_uid,user_uid,identity_source,token_version,refresh_hash,status,
|
|
|
+ rotated_from_uid,emergency_request_uid,created_at,last_seen_at,expires_at)
|
|
|
+ VALUES (CAST(:uid AS uuid),CAST(:family_uid AS uuid),
|
|
|
+ COALESCE(CAST(:identity_link_uid AS uuid),
|
|
|
+ (SELECT uid FROM public.enterprise_identity_links WHERE provider_uid=CAST(:provider_uid AS uuid)
|
|
|
+ AND enterprise_subject=:subject ORDER BY updated_at DESC LIMIT 1)),
|
|
|
+ CAST(:user_uid AS uuid),:identity_source,:token_version,
|
|
|
+ :refresh_hash,:status,CAST(:rotated_from_uid AS uuid),CAST(:emergency_request_uid AS uuid),
|
|
|
+ :created_at,:last_seen_at,:expires_at)
|
|
|
+ """), {**record, "identity_link_uid": record.get("identity_link_uid"),
|
|
|
+ "emergency_request_uid": record.get("emergency_request_uid")})
|
|
|
+
|
|
|
+ def put_session(self, record: dict[str, Any]) -> None:
|
|
|
+ self._insert_session(record)
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def create_session(self, record: dict[str, Any], *, max_sessions: int, commit: bool = True) -> None:
|
|
|
+ # Serialize all session creations for one account. Without this lock two
|
|
|
+ # concurrent transactions can both observe the same pre-insert count.
|
|
|
+ locked = self.session.execute(
|
|
|
+ text("SELECT id FROM public.users WHERE id=CAST(:user AS uuid) FOR UPDATE"),
|
|
|
+ {"user": record["user_uid"]},
|
|
|
+ ).scalar_one_or_none()
|
|
|
+ if not locked:
|
|
|
+ raise IdentityPolicyError("identity session user does not exist")
|
|
|
+ self._insert_session(record)
|
|
|
+ self.session.execute(text("""
|
|
|
+ WITH excess AS (
|
|
|
+ SELECT uid FROM public.identity_sessions WHERE user_uid=CAST(:user AS uuid) AND status='active'
|
|
|
+ ORDER BY created_at DESC,uid DESC OFFSET :maximum
|
|
|
+ ) UPDATE public.identity_sessions s SET status='revoked',revoke_reason='concurrency_limit',
|
|
|
+ revoked_at=CURRENT_TIMESTAMP FROM excess WHERE s.uid=excess.uid
|
|
|
+ """), {"user": record["user_uid"], "maximum": max_sessions})
|
|
|
+ if commit:
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def insert_exchange_code(self, record: dict[str, Any], *, commit: bool = True) -> None:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_exchange_codes(uid,code_hash,session_uid,redirect_uri,status,expires_at)
|
|
|
+ VALUES(CAST(:uid AS uuid),:code_hash,CAST(:session_uid AS uuid),:redirect_uri,'pending',:expires_at)
|
|
|
+ """), record)
|
|
|
+ if commit:
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def rotate_refresh(self, old_hash: str, new_session: dict[str, Any]) -> str:
|
|
|
+ row = self.session.execute(text("""
|
|
|
+ UPDATE public.identity_sessions SET status='rotated',rotated_at=:rotated_at
|
|
|
+ WHERE refresh_hash=:refresh_hash AND status='active' AND expires_at>:rotated_at
|
|
|
+ RETURNING family_uid::text
|
|
|
+ """), {"refresh_hash": old_hash, "rotated_at": new_session["created_at"]}).mappings().one_or_none()
|
|
|
+ if row:
|
|
|
+ self._insert_session(new_session)
|
|
|
+ self.session.commit()
|
|
|
+ return "rotated"
|
|
|
+ old = self.session.execute(text("SELECT family_uid::text,status FROM public.identity_sessions WHERE refresh_hash=:hash FOR UPDATE"),
|
|
|
+ {"hash": old_hash}).mappings().one_or_none()
|
|
|
+ if not old:
|
|
|
+ self.session.rollback()
|
|
|
+ return "unknown"
|
|
|
+ self.session.execute(text("UPDATE public.identity_sessions SET status='revoked',revoke_reason='refresh_reuse',revoked_at=CURRENT_TIMESTAMP WHERE family_uid=CAST(:family AS uuid)"),
|
|
|
+ {"family": old["family_uid"]})
|
|
|
+ self.session.commit()
|
|
|
+ return "reuse"
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _session(row: Any) -> dict[str, Any] | None:
|
|
|
+ if not row:
|
|
|
+ return None
|
|
|
+ result = dict(row)
|
|
|
+ for key in ("uid", "family_uid", "identity_link_uid", "user_uid", "rotated_from_uid", "emergency_request_uid"):
|
|
|
+ result[key] = str(result[key]) if result.get(key) else None
|
|
|
+ result.setdefault("roles", [])
|
|
|
+ result.setdefault("subject", result["user_uid"])
|
|
|
+ return result
|
|
|
+
|
|
|
+ def get_session(self, uid: str) -> dict[str, Any] | None:
|
|
|
+ row = self.session.execute(text("""
|
|
|
+ SELECT s.*,l.provider_uid::text provider_uid,COALESCE(l.enterprise_subject,s.user_uid::text) subject,
|
|
|
+ COALESCE(array_agg(r.name) FILTER (WHERE r.name IS NOT NULL),ARRAY[]::varchar[]) roles
|
|
|
+ FROM public.identity_sessions s LEFT JOIN public.user_roles ur ON ur.user_id=s.user_uid
|
|
|
+ LEFT JOIN public.roles r ON r.id=ur.role_id
|
|
|
+ LEFT JOIN public.enterprise_identity_links l ON l.uid=s.identity_link_uid
|
|
|
+ WHERE s.uid=CAST(:uid AS uuid) GROUP BY s.uid,l.provider_uid,l.enterprise_subject
|
|
|
+ """), {"uid": uid}).mappings().one_or_none()
|
|
|
+ return self._session(row)
|
|
|
+
|
|
|
+ def get_session_by_refresh(self, refresh_hash: str) -> dict[str, Any] | None:
|
|
|
+ uid = self.session.execute(text("SELECT uid::text FROM public.identity_sessions WHERE refresh_hash=:value"),
|
|
|
+ {"value": refresh_hash}).scalar_one_or_none()
|
|
|
+ return self.get_session(uid) if uid else None
|
|
|
+
|
|
|
+ def update_session(self, uid: str, *, commit: bool = True, **changes: Any) -> dict[str, Any]:
|
|
|
+ allowed = {"status", "revoke_reason", "risk_reason", "last_seen_at", "rotated_at", "revoked_at"}
|
|
|
+ safe = {key: value for key, value in changes.items() if key in allowed}
|
|
|
+ if safe:
|
|
|
+ assignments = ",".join(f"{key}=:{key}" for key in safe)
|
|
|
+ self.session.execute(text(f"UPDATE public.identity_sessions SET {assignments} WHERE uid=CAST(:uid AS uuid)"), {**safe, "uid": uid})
|
|
|
+ if commit:
|
|
|
+ self.session.commit()
|
|
|
+ return self.get_session(uid) or {}
|
|
|
+
|
|
|
+ def list_sessions(self, provider_uid: str | None = None, subject: str | None = None) -> list[dict[str, Any]]:
|
|
|
+ sql = "SELECT s.uid::text FROM public.identity_sessions s LEFT JOIN public.enterprise_identity_links l ON l.uid=s.identity_link_uid"
|
|
|
+ params: dict[str, Any] = {}
|
|
|
+ clauses: list[str] = []
|
|
|
+ if provider_uid:
|
|
|
+ clauses.append("l.provider_uid=CAST(:provider AS uuid)")
|
|
|
+ params["provider"] = provider_uid
|
|
|
+ if subject:
|
|
|
+ clauses.append("l.enterprise_subject=:subject")
|
|
|
+ params["subject"] = subject
|
|
|
+ if clauses:
|
|
|
+ sql += " WHERE " + " AND ".join(clauses)
|
|
|
+ uids = self.session.execute(text(sql), params).scalars().all()
|
|
|
+ return [record for uid in uids if (record := self.get_session(uid))]
|
|
|
+
|
|
|
+ def revoke_family(self, family_uid: str, *, reason: str) -> None:
|
|
|
+ self.session.execute(text("UPDATE public.identity_sessions SET status='revoked',revoke_reason=:reason,revoked_at=CURRENT_TIMESTAMP WHERE family_uid=CAST(:family AS uuid)"), {"family": family_uid, "reason": reason})
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def revoke_subject_sessions(self, provider_uid: str | None, subject: str, *, reason: str,
|
|
|
+ commit: bool = True) -> None:
|
|
|
+ if not provider_uid:
|
|
|
+ return
|
|
|
+ self.session.execute(text("""
|
|
|
+ UPDATE public.identity_sessions SET status='revoked',revoke_reason=:reason,revoked_at=CURRENT_TIMESTAMP
|
|
|
+ WHERE status='active' AND identity_link_uid IN (
|
|
|
+ SELECT uid FROM public.enterprise_identity_links
|
|
|
+ WHERE provider_uid=CAST(:provider AS uuid) AND enterprise_subject=:subject
|
|
|
+ )
|
|
|
+ """), {"provider": provider_uid, "subject": subject, "reason": reason})
|
|
|
+ if commit:
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def get_directory_event(self, provider_uid: str, source: str, source_event_id: str) -> dict[str, Any] | None:
|
|
|
+ row = self.session.execute(text("SELECT payload_digest,safe_result FROM public.identity_directory_events WHERE provider_uid=CAST(:provider AS uuid) AND source=:source AND source_event_id=:event"),
|
|
|
+ {"provider": provider_uid, "source": source, "event": source_event_id}).mappings().one_or_none()
|
|
|
+ return {"payload_digest": row["payload_digest"], "result": dict(row["safe_result"])} if row else None
|
|
|
+
|
|
|
+ def put_directory_event(self, event: dict[str, Any]) -> None:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_directory_events
|
|
|
+ (uid,provider_uid,source,source_event_id,cursor_value,cursor_sequence,event_type,enterprise_subject,payload_digest,status,safe_result,processed_at)
|
|
|
+ VALUES (CAST(:uid AS uuid),CAST(:provider_uid AS uuid),:source,:source_event_id,:cursor,:cursor_sequence,
|
|
|
+ :event_type,:subject,:payload_digest,'applied',CAST(:result AS jsonb),:processed_at)
|
|
|
+ """), {**event, "uid": new_governance_uid(), "event_type": event.get("event_type", "MOVER"),
|
|
|
+ "subject": event.get("subject"), "result": json.dumps(event["result"])})
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_directory_checkpoints(provider_uid,source,cursor_value,cursor_sequence,updated_at)
|
|
|
+ VALUES(CAST(:provider_uid AS uuid),:source,:cursor,:cursor_sequence,CURRENT_TIMESTAMP)
|
|
|
+ ON CONFLICT(provider_uid,source) DO UPDATE SET cursor_value=EXCLUDED.cursor_value,
|
|
|
+ cursor_sequence=EXCLUDED.cursor_sequence,updated_at=CURRENT_TIMESTAMP
|
|
|
+ """), event)
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def get_directory_checkpoint(self, provider_uid: str, source: str) -> dict[str, Any] | None:
|
|
|
+ row = self.session.execute(text("SELECT cursor_value cursor,cursor_sequence FROM public.identity_directory_checkpoints WHERE provider_uid=CAST(:provider AS uuid) AND source=:source"),
|
|
|
+ {"provider": provider_uid, "source": source}).mappings().one_or_none()
|
|
|
+ return dict(row) if row else None
|
|
|
+
|
|
|
+ def record_directory_conflict(self, *, commit: bool = True, **record: Any) -> None:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_directory_conflicts
|
|
|
+ (uid,provider_uid,source,source_event_id,cursor_value,cursor_sequence,payload_digest,reason)
|
|
|
+ VALUES(CAST(:uid AS uuid),CAST(:provider_uid AS uuid),:source,:source_event_id,:cursor,:cursor_sequence,:payload_digest,:reason)
|
|
|
+ """), {**record, "uid": new_governance_uid()})
|
|
|
+ if commit:
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def get_identity(self, provider_uid: str, subject: str, *, for_update: bool = False) -> dict[str, Any] | None:
|
|
|
+ lock = " FOR UPDATE" if for_update else ""
|
|
|
+ row = self.session.execute(text("""
|
|
|
+ SELECT user_uid::text,username,display_name,department_external_id department,groups,
|
|
|
+ role_names roles,authorization_scope,mapping_version,claims_digest,status,token_version
|
|
|
+ FROM public.enterprise_identity_links WHERE provider_uid=CAST(:provider AS uuid) AND enterprise_subject=:subject
|
|
|
+ """ + lock), {"provider": provider_uid, "subject": subject}).mappings().one_or_none()
|
|
|
+ return dict(row) if row else None
|
|
|
+
|
|
|
+ def put_identity(self, provider_uid: str, subject: str, identity: dict[str, Any], *, commit: bool = True) -> None:
|
|
|
+ existing = self.get_identity(provider_uid, subject, for_update=not commit)
|
|
|
+ if not existing:
|
|
|
+ if not all(identity.get(key) for key in ("provider_uid", "username", "display_name", "mapping_version", "claims_digest")):
|
|
|
+ raise IdentityPolicyError("JOINER requires an approved claims mapping result")
|
|
|
+ from app.core.system.auth import hash_password
|
|
|
+
|
|
|
+ self.session.execute(text("INSERT INTO public.users(id,username,display_name,password_hash,status) VALUES (CAST(:uid AS uuid),:username,:display,:password,'active')"),
|
|
|
+ {"uid": identity["user_uid"], "username": identity["username"],
|
|
|
+ "display": identity["display_name"], "password": hash_password(secrets.token_urlsafe(48))})
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.enterprise_identity_links
|
|
|
+ (uid,provider_uid,enterprise_subject,user_uid,username,display_name,department_external_id,groups,role_names,
|
|
|
+ authorization_scope,mapping_version,claims_digest,status,token_version)
|
|
|
+ VALUES (CAST(:uid AS uuid),CAST(:provider AS uuid),:subject,CAST(:user AS uuid),:username,:display,
|
|
|
+ COALESCE(:department,''),CAST(:groups AS jsonb),CAST(:roles AS jsonb),CAST(:scope AS jsonb),:mapping,
|
|
|
+ :digest,:status,:token_version)
|
|
|
+ """), {"uid": new_governance_uid(), "provider": identity["provider_uid"], "subject": subject,
|
|
|
+ "user": identity["user_uid"], "username": identity["username"], "display": identity["display_name"],
|
|
|
+ "department": identity.get("department"), "groups": json.dumps(identity.get("groups", [])),
|
|
|
+ "roles": json.dumps(identity.get("roles", [])), "scope": json.dumps(identity.get("authorization_scope", {})),
|
|
|
+ "mapping": identity["mapping_version"], "digest": identity["claims_digest"],
|
|
|
+ "status": identity["status"], "token_version": identity["token_version"]})
|
|
|
+ else:
|
|
|
+ self.session.execute(text("""
|
|
|
+ UPDATE public.enterprise_identity_links SET username=:username,department_external_id=COALESCE(:department,''),
|
|
|
+ display_name=COALESCE(:display_name,display_name),groups=CAST(:groups AS jsonb),role_names=CAST(:roles AS jsonb),
|
|
|
+ authorization_scope=CAST(:scope AS jsonb),mapping_version=COALESCE(:mapping_version,mapping_version),
|
|
|
+ claims_digest=COALESCE(:claims_digest,claims_digest),status=:status,token_version=:token_version,
|
|
|
+ updated_at=CURRENT_TIMESTAMP WHERE provider_uid=CAST(:provider AS uuid) AND enterprise_subject=:subject
|
|
|
+ """), {"provider": provider_uid, "subject": subject, "username": identity["username"], "department": identity.get("department"),
|
|
|
+ "display_name": identity.get("display_name"), "groups": json.dumps(identity.get("groups", [])),
|
|
|
+ "roles": json.dumps(identity.get("roles", [])), "scope": json.dumps(identity.get("authorization_scope", {})),
|
|
|
+ "mapping_version": identity.get("mapping_version"), "claims_digest": identity.get("claims_digest"),
|
|
|
+ "status": identity["status"], "token_version": identity["token_version"]})
|
|
|
+ self.session.execute(text("DELETE FROM public.user_roles WHERE user_id=CAST(:user AS uuid)"), {"user": identity["user_uid"]})
|
|
|
+ self.session.execute(text("INSERT INTO public.user_roles(user_id,role_id) SELECT CAST(:user AS uuid),id FROM public.roles WHERE name=ANY(:roles)"),
|
|
|
+ {"user": identity["user_uid"], "roles": identity.get("roles", [])})
|
|
|
+ self.session.execute(
|
|
|
+ text("UPDATE public.users SET status=:status,updated_at=CURRENT_TIMESTAMP WHERE id=CAST(:user AS uuid)"),
|
|
|
+ {"user": identity["user_uid"], "status": "active" if identity["status"] == "active" else "disabled"},
|
|
|
+ )
|
|
|
+ if commit:
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def get_organization_node(self, provider_uid: str, node_type: str, external_id: str,
|
|
|
+ *, for_update: bool = False) -> dict[str, Any] | None:
|
|
|
+ lock = " FOR UPDATE" if for_update else ""
|
|
|
+ row = self.session.execute(text("""
|
|
|
+ SELECT uid::text,provider_uid::text,external_id,node_type,parent_external_id,display_name,status,attributes,updated_at
|
|
|
+ FROM public.identity_organization_nodes
|
|
|
+ WHERE provider_uid=CAST(:provider AS uuid) AND node_type=:node_type AND external_id=:external_id
|
|
|
+ """ + lock), {"provider": provider_uid, "node_type": node_type,
|
|
|
+ "external_id": external_id}).mappings().one_or_none()
|
|
|
+ return dict(row) if row else None
|
|
|
+
|
|
|
+ def put_organization_node(self, node: dict[str, Any], *, commit: bool = True) -> None:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_organization_nodes
|
|
|
+ (uid,provider_uid,external_id,node_type,parent_external_id,display_name,status,attributes,updated_at)
|
|
|
+ VALUES(CAST(:uid AS uuid),CAST(:provider_uid AS uuid),:external_id,:node_type,:parent_external_id,
|
|
|
+ :display_name,:status,CAST(:attributes AS jsonb),:updated_at)
|
|
|
+ ON CONFLICT(provider_uid,node_type,external_id) DO UPDATE SET
|
|
|
+ parent_external_id=EXCLUDED.parent_external_id,display_name=EXCLUDED.display_name,
|
|
|
+ status=EXCLUDED.status,attributes=EXCLUDED.attributes,updated_at=EXCLUDED.updated_at
|
|
|
+ """), {**node, "attributes": json.dumps(node.get("attributes", {}))})
|
|
|
+ if commit:
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def _insert_directory_event(self, event: dict[str, Any], result: dict[str, Any]) -> None:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_directory_events
|
|
|
+ (uid,provider_uid,source,source_event_id,cursor_value,cursor_sequence,event_type,enterprise_subject,payload_digest,status,safe_result,processed_at)
|
|
|
+ VALUES (CAST(:uid AS uuid),CAST(:provider_uid AS uuid),:source,:source_event_id,:cursor,:cursor_sequence,
|
|
|
+ :event_type,:subject,:payload_digest,'applied',CAST(:result AS jsonb),:processed_at)
|
|
|
+ """), {**event, "uid": new_governance_uid(), "subject": event.get("subject"),
|
|
|
+ "result": json.dumps(result)})
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_directory_checkpoints(provider_uid,source,cursor_value,cursor_sequence,updated_at)
|
|
|
+ VALUES(CAST(:provider_uid AS uuid),:source,:cursor,:cursor_sequence,CURRENT_TIMESTAMP)
|
|
|
+ ON CONFLICT(provider_uid,source) DO UPDATE SET cursor_value=EXCLUDED.cursor_value,
|
|
|
+ cursor_sequence=EXCLUDED.cursor_sequence,updated_at=CURRENT_TIMESTAMP
|
|
|
+ """), event)
|
|
|
+
|
|
|
+ def apply_directory_event_atomic(self, event: dict[str, Any],
|
|
|
+ mutation: Callable[[], dict[str, Any]]) -> dict[str, Any]:
|
|
|
+ """Serialize one provider/source feed and commit its complete effect once."""
|
|
|
+ try:
|
|
|
+ scope = f"{event['provider_uid']}:{event['source']}"
|
|
|
+ self.session.execute(text("SELECT pg_advisory_xact_lock(hashtextextended(:scope,0))"), {"scope": scope})
|
|
|
+ previous = self.get_directory_event(event["provider_uid"], event["source"], event["source_event_id"])
|
|
|
+ if previous:
|
|
|
+ if previous["payload_digest"] != event["payload_digest"]:
|
|
|
+ self.record_directory_conflict(commit=False, **event, reason="source_event_id_payload_mismatch")
|
|
|
+ self.session.commit()
|
|
|
+ raise IdentityPolicyError("directory event idempotency conflict")
|
|
|
+ self.session.commit()
|
|
|
+ return {**previous["result"], "idempotent": True}
|
|
|
+ checkpoint = self.get_directory_checkpoint(event["provider_uid"], event["source"])
|
|
|
+ if checkpoint and event["cursor_sequence"] <= checkpoint["cursor_sequence"]:
|
|
|
+ self.record_directory_conflict(commit=False, **event, reason="cursor_not_monotonic")
|
|
|
+ self.session.commit()
|
|
|
+ raise IdentityPolicyError("directory cursor is not monotonic")
|
|
|
+ result = mutation()
|
|
|
+ self._insert_directory_event(event, result)
|
|
|
+ self.session.commit()
|
|
|
+ return result
|
|
|
+ except IdentityPolicyError:
|
|
|
+ # Policy conflicts are committed above so operators retain evidence;
|
|
|
+ # mutation failures are rolled back with every partial lifecycle write.
|
|
|
+ if self.session.in_transaction():
|
|
|
+ self.session.rollback()
|
|
|
+ raise
|
|
|
+ except Exception:
|
|
|
+ self.session.rollback()
|
|
|
+ raise
|
|
|
+
|
|
|
+ def put_emergency(self, record: dict[str, Any]) -> None:
|
|
|
+ existing = self.get_emergency(record["uid"])
|
|
|
+ if not existing:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_emergency_requests
|
|
|
+ (uid,requester_uid,emergency_account_uid,reason,status,starts_at,expires_at)
|
|
|
+ VALUES (CAST(:uid AS uuid),CAST(:requester_uid AS uuid),CAST(:account_uid AS uuid),:reason,:status,:starts_at,:expires_at)
|
|
|
+ """), {**record, "starts_at": record["requested_at"]})
|
|
|
+ else:
|
|
|
+ self.session.execute(text("UPDATE public.identity_emergency_requests SET status=:status,activated_at=:activated_at,closed_at=:closed_at,reviewer_uid=CAST(:reviewer_uid AS uuid),review_outcome=:review_outcome WHERE uid=CAST(:uid AS uuid)"),
|
|
|
+ {"uid": record["uid"], "status": record["status"], "activated_at": record.get("activated_at"),
|
|
|
+ "closed_at": record.get("closed_at"), "reviewer_uid": record.get("reviewer_uid"),
|
|
|
+ "review_outcome": record.get("review_outcome")})
|
|
|
+ for approver in set(record.get("approver_uids", [])) - set(existing.get("approver_uids", [])):
|
|
|
+ self.session.execute(text("INSERT INTO public.identity_emergency_approvals(request_uid,approver_uid) VALUES (CAST(:request AS uuid),CAST(:approver AS uuid)) ON CONFLICT DO NOTHING"), {"request": record["uid"], "approver": approver})
|
|
|
+ if record["status"] in {"closed", "expired", "reviewed"}:
|
|
|
+ self.session.execute(text("""
|
|
|
+ UPDATE public.identity_sessions SET status='revoked',revoke_reason=:reason,revoked_at=CURRENT_TIMESTAMP
|
|
|
+ WHERE emergency_request_uid=CAST(:request AS uuid) AND status='active'
|
|
|
+ """), {"request": record["uid"], "reason": f"emergency_{record['status']}"})
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def get_emergency(self, uid: str) -> dict[str, Any] | None:
|
|
|
+ row = self.session.execute(text("""
|
|
|
+ SELECT r.uid::text,r.requester_uid::text,r.emergency_account_uid::text account_uid,r.reason,r.status,
|
|
|
+ r.starts_at requested_at,r.expires_at,r.activated_at,r.reviewer_uid::text,r.review_outcome,
|
|
|
+ COALESCE(array_agg(a.approver_uid::text) FILTER (WHERE a.approver_uid IS NOT NULL),ARRAY[]::text[]) approver_uids
|
|
|
+ FROM public.identity_emergency_requests r LEFT JOIN public.identity_emergency_approvals a ON a.request_uid=r.uid
|
|
|
+ WHERE r.uid=CAST(:uid AS uuid) GROUP BY r.uid
|
|
|
+ """), {"uid": uid}).mappings().one_or_none()
|
|
|
+ return dict(row) if row else None
|
|
|
+
|
|
|
+ def audit(self, event: dict[str, Any]) -> None:
|
|
|
+ self.session.execute(text("""
|
|
|
+ INSERT INTO public.identity_audit_events
|
|
|
+ (uid,event_type,outcome,actor_uid,provider_uid,enterprise_subject_digest,resource_type,resource_uid,correlation_uid,safe_detail)
|
|
|
+ VALUES (CAST(:uid AS uuid),:event_type,:outcome,CAST(:actor_uid AS uuid),CAST(:provider_uid AS uuid),:subject_digest,
|
|
|
+ :resource_type,:resource_uid,CAST(:correlation_uid AS uuid),CAST(:safe_detail AS jsonb))
|
|
|
+ """), {"uid": new_governance_uid(), "event_type": event["event_type"], "outcome": event["outcome"],
|
|
|
+ "actor_uid": event.get("actor_uid"), "provider_uid": event.get("provider_uid"),
|
|
|
+ "subject_digest": event.get("subject_digest"), "resource_type": event.get("resource_type", "identity"),
|
|
|
+ "resource_uid": event.get("resource_uid"), "correlation_uid": event.get("correlation_uid"),
|
|
|
+ "safe_detail": json.dumps(event.get("safe_detail", {}))})
|
|
|
+ self.session.commit()
|
|
|
+
|
|
|
+ def emergency_session_active(self, request_uid: str, account_uid: str, now: Any) -> bool:
|
|
|
+ return bool(self.session.execute(text("""
|
|
|
+ SELECT 1 FROM public.identity_emergency_requests e
|
|
|
+ JOIN public.users u ON u.id=e.emergency_account_uid
|
|
|
+ WHERE e.uid=CAST(:request AS uuid) AND e.emergency_account_uid=CAST(:account AS uuid)
|
|
|
+ AND e.status='active' AND e.starts_at<=:now AND e.expires_at>:now AND u.status='active'
|
|
|
+ AND EXISTS (
|
|
|
+ SELECT 1 FROM public.user_roles ur JOIN public.roles r ON r.id=ur.role_id
|
|
|
+ WHERE ur.user_id=u.id AND r.name='admin'
|
|
|
+ )
|
|
|
+ AND NOT EXISTS (
|
|
|
+ SELECT 1 FROM public.enterprise_identity_links l WHERE l.user_uid=u.id
|
|
|
+ )
|
|
|
+ """), {"request": request_uid, "account": account_uid, "now": now}).scalar())
|