from __future__ import annotations from collections.abc import Mapping from datetime import datetime import hashlib import json from urllib.parse import parse_qs, quote, urlsplit from sqlalchemy.orm import Session from govoplan_core.core.institutional import EvidenceReference from govoplan_cases.backend.domain import DEFAULT_CASE_ACCESS_PURPOSE, CaseRecord from govoplan_cases.backend.service import CaseStoreError, get_case, update_case _SOURCE_CONTRACT = "cases.quick-access-reference.v1" _SUPPORTED_REFERENCES = { ("files", "file-version"): "document", ("mail", "message"): "event", ("campaigns", "campaign"): "event", } def link_case_evidence( session: Session, principal: object, *, case_id: str, expected_revision: int, reference: Mapping[str, object], recorded_at: datetime, change_reason: str, idempotency_key: str, purpose: str = DEFAULT_CASE_ACCESS_PURPOSE, ) -> CaseRecord: base = _base_revision( session, principal, case_id=case_id, expected_revision=expected_revision, purpose=purpose, ) evidence = _validated_evidence(reference, tenant_id=base.reference.tenant_id) if any(item.evidence_id == evidence.evidence_id for item in base.evidence_refs): raise CaseStoreError("The owner reference is already linked to this Case.") return update_case( session, principal, case_id=case_id, expected_revision=expected_revision, changes={"evidence_refs": (*base.evidence_refs, evidence)}, recorded_at=recorded_at, change_reason=change_reason, idempotency_key=idempotency_key, purpose=purpose, ) def unlink_case_evidence( session: Session, principal: object, *, case_id: str, evidence_id: str, expected_revision: int, recorded_at: datetime, change_reason: str, idempotency_key: str, purpose: str = DEFAULT_CASE_ACCESS_PURPOSE, ) -> CaseRecord: base = _base_revision( session, principal, case_id=case_id, expected_revision=expected_revision, purpose=purpose, ) evidence = next( (item for item in base.evidence_refs if item.evidence_id == evidence_id), None, ) if evidence is None or linked_source(evidence) is None: raise CaseStoreError("The linked owner reference was not found.") return update_case( session, principal, case_id=case_id, expected_revision=expected_revision, changes={ "evidence_refs": tuple( item for item in base.evidence_refs if item.evidence_id != evidence_id ) }, recorded_at=recorded_at, change_reason=change_reason, idempotency_key=idempotency_key, purpose=purpose, ) def linked_source(evidence: EvidenceReference) -> Mapping[str, str] | None: if evidence.source_ref is None: return None try: value = json.loads(evidence.source_ref) except (TypeError, ValueError): return None if not isinstance(value, Mapping) or value.get("contract") != _SOURCE_CONTRACT: return None result = { key: str(value.get(key) or "") for key in ("owner_module", "kind", "object_id", "path") } if ( not all(result.values()) or (result["owner_module"], result["kind"]) not in _SUPPORTED_REFERENCES ): return None return result def _base_revision( session: Session, principal: object, *, case_id: str, expected_revision: int, purpose: str, ) -> CaseRecord: base = get_case( session, principal, case_id=case_id, revision=expected_revision, purpose=purpose, ) if base is None: raise LookupError("Case revision not found.") return base def _validated_evidence( value: Mapping[str, object], *, tenant_id: str, ) -> EvidenceReference: owner_module = _bounded(value.get("owner_module"), "Owner module", 120) reference_kind = _bounded(value.get("kind"), "Owner reference kind", 120) object_id = _bounded(value.get("object_id"), "Owner object id", 1_000) reference_tenant = _bounded(value.get("tenant_id"), "Owner tenant id", 255) version = _bounded(value.get("version"), "Owner version", 255) path = _bounded(value.get("path"), "Owner path", 2_000) evidence_kind = _SUPPORTED_REFERENCES.get((owner_module, reference_kind)) if evidence_kind is None: raise CaseStoreError("The owner reference kind is not supported by Cases.") if reference_tenant != tenant_id: raise CaseStoreError("Owner references cannot cross Case tenants.") _validate_owner_path( owner_module=owner_module, reference_kind=reference_kind, object_id=object_id, version=version, path=path, ) canonical = { "contract": _SOURCE_CONTRACT, "owner_module": owner_module, "kind": reference_kind, "object_id": object_id, "path": path, } encoded = json.dumps( canonical, ensure_ascii=True, sort_keys=True, separators=(",", ":"), ) digest = hashlib.sha256( f"{tenant_id}\x1f{encoded}\x1f{version}".encode() ).hexdigest() return EvidenceReference( kind=evidence_kind, # type: ignore[arg-type] owner_module=owner_module, evidence_id=f"linked-{digest[:40]}", tenant_id=tenant_id, version=version, source_ref=encoded, derived_from=(_SOURCE_CONTRACT,), ) def _validate_owner_path( *, owner_module: str, reference_kind: str, object_id: str, version: str, path: str, ) -> None: parsed = urlsplit(path) if parsed.scheme or parsed.netloc or not path.startswith("/") or path.startswith("//"): raise CaseStoreError("Owner paths must be local platform routes.") if (owner_module, reference_kind) == ("files", "file-version"): query = parse_qs(parsed.query, keep_blank_values=True) if ( parsed.path != "/files" or query.get("versionId") != [object_id] or version != object_id or not query.get("fileId", [""])[0] ): raise CaseStoreError("Files links require an exact authorized version path.") return if (owner_module, reference_kind) == ("mail", "message"): query = parse_qs(parsed.query, keep_blank_values=True) profile = query.get("profile", [""])[0] folder = query.get("folder", [""])[0] message = query.get("message", [""])[0] expected_id = f"{profile}:{folder}:{message}" if ( parsed.path != "/mail" or not profile or not folder or not message or object_id != expected_id or version != message ): raise CaseStoreError("Mail links require an exact mailbox message path.") return expected_path = f"/campaigns/{quote(object_id, safe='')}" if path != expected_path: raise CaseStoreError("Campaign links require an exact Campaign path.") def _bounded(value: object, label: str, maximum: int) -> str: result = str(value or "").strip() if not result: raise CaseStoreError(f"{label} is required.") if len(result) > maximum or any(ord(character) < 32 for character in result): raise CaseStoreError(f"{label} is invalid.") return result __all__ = [ "link_case_evidence", "linked_source", "unlink_case_evidence", ]