"""Reading contracts: the list, one contract, its versions, the certificate and the evidence check, the key-date
register, the status and the executed copy — every one deciding visibility from `caller` (support.View).
A body route (versions, certificate, evidence, executed copy) of an EMPLOYMENT contract answers 404 to anyone
without contract:employment:read — its owner and a read_all manager included — while show_contract still shows them
its existence, state and key dates (SC-CT-20).
"""
import base64
from typing import Literal, Optional
from sqlalchemy import and_, or_
from erp_platform.types import Limit
from brick_contracts.application import support as S
from brick_contracts.application.use_cases import CT
from brick_contracts.domain.services import CLAIM, CertificateBuilder, ChainVerifier
from brick_contracts.domain.values import EMPLOYMENT, EXECUTED, INTERNAL, STATES, sha256_hex
from brick_contracts.infrastructure import public_door
from brick_contracts.infrastructure import tables as T
def uow():
from brick_contracts.application import use_cases
return use_cases.repository()
def _visible_where(view: S.View):
"""Visibility as SQL, so the list is one set-based SELECT (hr filtered in Python after loading everyone)."""
c = T.contract.c
if S.has(view.caller, S.READ_ALL):
return None
conds = []
if S.has(view.caller, S.EMPLOYMENT_READ):
conds.append(c.sensitivity == EMPLOYMENT)
if S.has(view.caller, S.READ) or S.has(view.caller, S.WRITE):
conds.append(c.owner == view.me)
if view.groups:
conds.append(c.owner_group.in_(sorted(view.groups)))
extra = sorted(view.approving | view.signing)
if extra:
conds.append(c.contract_id.in_(extra))
return or_(*conds) if conds else c.contract_id == "∅"
@CT.use_case(kind="query", route="GET /contracts")
def list_contracts(caller: dict, state: Optional[Literal[STATES]] = None, type_code: Optional[S.Code] = None,
mine: bool = False, limit: Limit = 200) -> dict:
"""The contracts you may see: state (ACTIVE / EXPIRED derived from the term), counterparty, type, owner and the
next key date. Never a contract's body."""
with uow() as u:
view = S.View(u, caller)
c = T.contract.c
conds = [w for w in (_visible_where(view),) if w is not None]
if type_code:
conds.append(c.type_code == type_code.lower())
if mine:
conds.append(c.owner == view.me)
if state and state not in ("ACTIVE", "EXPIRED"):
conds.append(c.state == state)
rows = u.contracts.headers(where=and_(*conds) if conds else None, limit=limit)
nxt = u.contracts.next_key_dates([r["contract_id"] for r in rows])
out = []
for r in rows:
derived = r["state"]
if r["state"] in EXECUTED:
ends = S.today() > r["ends_on"] if r.get("ends_on") else False
starts = r.get("starts_on")
derived = "EXPIRED" if ends else ("ACTIVE" if starts is None or starts <= S.today() else r["state"])
if state in ("ACTIVE", "EXPIRED") and derived != state:
continue
k = nxt.get(r["contract_id"])
out.append({"contract_id": r["contract_id"], "title": r["title"], "type_code": r["type_code"],
"state": derived, "bp_id": r["bp_id"], "owner": r["owner"], "owner_group": r["owner_group"],
"sensitivity": r["sensitivity"], "version": r["current_version"] or None,
"next_key_date": {"kind": k["kind"], "due_on": k["due_on"].isoformat()} if k else None})
return {"contracts": out}
@CT.use_case(kind="query", route="GET /contracts/{contract_id}", etag="row_version")
def show_contract(caller: dict, contract_id: S.ContractId) -> dict:
"""One contract. With its body — sections (position, deviation level, overridden), variables with their source,
signers (seen apart from signed, whose turn), approvals, versions, the readiness panel, open manual tasks — when
you may see it; an employment contract's body needs contract:employment:read, and without it you see its state and
key dates only. Send `row_version` back as If-Match / expected_version with an edit."""
with uow() as u:
view = S.View(u, caller)
c = u.contracts.get(contract_id)
if c is None or not view.sees(c):
raise S.not_found(f"contract {contract_id}")
if not view.body(c):
return S.summary(c, view.today)
links = {}
for s in c.signers:
if s.kind == "EXTERNAL":
st = [x for x in public_door.status(c.contract_id, s.signer_no) if x["grant"] == "SIGN"]
if st:
last = st[-1]
links[s.signer_no] = {"state": last["state"], "seen_count": last["seen_count"],
"first_seen_at": S.iso(last["first_seen_at"]),
"last_seen_at": S.iso(last["last_seen_at"]), "used_at": S.iso(last["used_at"]),
"expires_at": S.iso(last["expires_at"])}
return S.full(c, view, links=links)
def _body(view, contract_id: str, *, signer_ok: bool = False):
c = view.uow.contracts.get(contract_id)
if c is None:
raise S.not_found(f"contract {contract_id}")
if view.body(c):
return c
if signer_ok and any(s.kind == INTERNAL and s.is_me(view.caller) for s in c.signers) and S.has(view.caller, S.SIGN):
return c
raise S.not_found(f"contract {contract_id}")
@CT.use_case(kind="query", route="GET /contracts/{contract_id}/versions")
def show_versions(caller: dict, contract_id: S.ContractId) -> dict:
"""The versions sent (ISSUED / SUPERSEDED / SIGNED): when, by whom, the sha256 and the frozen words of each. A sent
version never changes; a re-send is the next number."""
with uow() as u:
view = S.View(u, caller)
c = _body(view, contract_id)
return {"contract_id": c.contract_id, "versions": [
{"version": v.version, "status": v.status, "sha256": v.sha256, "size_bytes": v.size_bytes,
"frozen_at": S.iso(v.frozen_at), "frozen_by": v.frozen_by, "draft_fingerprint": v.draft_fingerprint,
"text": v.text_body} for v in sorted(c.versions.values(), key=lambda v: v.version)]}
@CT.use_case(kind="query", route="GET /contracts/{contract_id}/versions/{version}/file")
def download_version(caller: dict, contract_id: S.ContractId, version: S.VersionNo) -> dict:
"""A frozen version's PDF, streamed as base64 with its sha256 (checked against the stored bytes) — never a link.
Our signatory may download the version they are asked to sign."""
with uow() as u:
view = S.View(u, caller)
c = _body(view, contract_id, signer_ok=True)
v = c.versions.get(int(version))
if v is None:
raise S.not_found(f"version {version} of {contract_id}")
got = u.objects.get(v.object_key)
if got is None or sha256_hex(got[0]) != v.sha256:
raise S.Refused(409, f"version {version}'s stored bytes do not match its sha256 — it is not released")
return {"contract_id": c.contract_id, "version": v.version, "filename": f"{c.contract_id}-v{v.version}.pdf",
"content_type": "application/pdf", "sha256": v.sha256, "size": len(got[0]),
"content_base64": base64.b64encode(got[0]).decode()}
def _certificate(c, cfg) -> dict:
chain = ChainVerifier.verify(c.contract_id, c.events)
signed = c.versions.get(c.executed.signed_version) if c.executed and c.executed.signed_version else c.current
lines = CertificateBuilder.lines(contract_id=c.contract_id, title=c.title, version=signed, signers=c.signers,
events=c.events, chain=chain, retention_note=S.retention_note(cfg, c)) if signed else []
return {"contract_id": c.contract_id, "claim": CLAIM, "chain": {"valid": chain[0], "broken_at": chain[1]},
"version": signed.version if signed else None, "version_sha256": signed.sha256 if signed else None,
"executed_sha256": c.executed.sha256 if c.executed else None,
"executed_provenance": c.executed.provenance if c.executed else None,
"signers": [S.signer_json(s) for s in c.signers],
"events": [{"seq": e.seq, "kind": e.kind, "at": S.iso(e.occurred_at), "signer_no": e.signer_no,
"signer_kind": e.signer_kind, "version": e.version, "sha256_shown": e.version_sha256_shown,
"email": e.email, "ip": e.ip, "user_agent": e.user_agent, "typed_name": e.typed_name,
"auth_method": e.auth_method, "pool_sub": e.pool_sub, "token_ref": e.token_ref,
"consent_text_version": e.consent_text_version, "detail": e.detail,
"row_hash": e.row_hash} for e in c.events],
"text": "\n".join(lines)}
@CT.use_case(kind="query", route="GET /contracts/{contract_id}/certificate", permission=S.EVIDENCE)
def show_certificate(caller: dict, contract_id: S.ContractId) -> dict:
"""The certificate (CT-13): every signer (name, email, side, title, authority), every event with its time, IP and
user agent, the sha256 of the signed version and of the executed file, and whether the event chain verifies
(recomputed on every read; a broken chain names the first event that does not match). It says "signed
electronically" — never advanced, qualified or anything more."""
with uow() as u:
view = S.View(u, caller)
c = _body(view, contract_id)
return _certificate(c, view.config)
@CT.use_case(kind="query", route="GET /contracts/{contract_id}/evidence", permission=S.EVIDENCE)
def verify_evidence(caller: dict, contract_id: S.ContractId) -> dict:
"""Check a contract's evidence (CT-32): the hash chain recomputed event by event, and the stored bytes of every
version and of the executed copy against their sha256. `valid` is false — naming where — if any row or file was
changed."""
with uow() as u:
view = S.View(u, caller)
c = _body(view, contract_id)
ok, broken = ChainVerifier.verify(c.contract_id, c.events)
files = []
for v in sorted(c.versions.values(), key=lambda v: v.version):
got = u.objects.get(v.object_key)
files.append({"what": f"version {v.version}", "sha256": v.sha256,
"matches": got is not None and sha256_hex(got[0]) == v.sha256})
if c.executed:
got = u.objects.get(c.executed.object_key)
files.append({"what": "executed copy", "sha256": c.executed.sha256,
"matches": got is not None and sha256_hex(got[0]) == c.executed.sha256})
return {"contract_id": c.contract_id, "valid": ok and all(f["matches"] for f in files),
"chain": {"valid": ok, "broken_at": broken, "events": len(c.events)}, "files": files}
@CT.use_case(kind="query", route="GET /key-dates")
def key_dates(caller: dict, before: Optional[S.Day] = None, kind: Optional[S.Code] = None, limit: Limit = 500) -> dict:
"""The key-date register (CT-15): open END, NOTICE_DEADLINE (end − notice, its own date), PROBATION_END,
PRINCIPAL_STATEMENT_DUE … with their owner and the next reminder, for the contracts you may see."""
from datetime import date
with uow() as u:
view = S.View(u, caller)
w = _visible_where(view)
heads = {h["contract_id"]: h for h in u.contracts.headers(where=w, limit=5000)}
rows = u.contracts.register(sorted(heads))
out = []
for r in rows:
d = r["due_on"] if isinstance(r["due_on"], date) else date.fromisoformat(str(r["due_on"])[:10])
if before and d > before:
continue
if kind and r["kind"] != kind.upper():
continue
h = heads[r["contract_id"]]
out.append({"contract_id": r["contract_id"], "title": h["title"], "kind": r["kind"], "seq": r["seq"],
"due_on": d.isoformat(), "owner_group": r["owner_group"],
"next_reminder_on": S.iso(r.get("next_remind_on"))})
if len(out) >= limit:
break
return {"key_dates": out}
def status_of(c, view, *, executed: bool) -> dict:
v = c.current
return {"contract_id": c.contract_id, "state": c.derived_state(view.today), "version": c.current_version or None,
"signers": [{"signer_no": s.signer_no, "side": s.side, "status": s.status,
"seen": s.first_opened_at is not None} for s in c.signers],
"executed": ({"object_key": c.executed.object_key, "sha256": c.executed.sha256,
"provenance": c.executed.provenance} if executed and c.executed else None),
"key_dates": [{"kind": k.kind, "due_on": k.due_on.isoformat(), "done": k.done_at is not None}
for k in c.key_dates],
"principal_statement_issued_at": S.iso(c.principal_statement_issued_at),
"warnings": S.warnings(view.config, c, view.today)}
@CT.use_case(kind="query", route="GET /contracts/{contract_id}/status")
def contract_status(caller: dict, contract_id: S.ContractId) -> dict:
"""Where a contract stands: its state, version, each signer seen and acted, the key dates, when the written
statement was given and any warning — the same answer contracts.status gives a calling module."""
with uow() as u:
view = S.View(u, caller)
c = u.contracts.get(contract_id)
if c is None or not view.sees(c):
raise S.not_found(f"contract {contract_id}")
return status_of(c, view, executed=view.body(c))
@CT.use_case(kind="query", route="GET /contracts/{contract_id}/executed")
def fetch_executed(caller: dict, contract_id: S.ContractId) -> dict:
"""The executed copy, streamed as base64 with its sha256 — never an object link — after the evidence chain and the
file's sha256 are checked (a tampered contract's copy is refused). Our signatory on it may download it."""
with uow() as u:
view = S.View(u, caller)
c = _body(view, contract_id, signer_ok=True)
return executed_bytes(u, c)
def executed_bytes(u, c) -> dict:
if c.executed is None:
raise S.Refused(409, "there is no executed copy yet")
ok, broken = ChainVerifier.verify(c.contract_id, c.events)
got = u.objects.get(c.executed.object_key)
if not ok:
raise S.Refused(409, f"the evidence chain is broken at event {broken}; the executed copy is not released")
if got is None or sha256_hex(got[0]) != c.executed.sha256:
raise S.Refused(409, "the executed copy's bytes no longer match its sha256; it is not released")
return {"contract_id": c.contract_id, "filename": c.executed.filename or f"{c.contract_id}.pdf",
"content_type": c.executed.content_type or "application/pdf", "sha256": c.executed.sha256,
"size": len(got[0]), "content_base64": base64.b64encode(got[0]).decode()}
__all__ = ["list_contracts", "show_contract", "show_versions", "download_version", "show_certificate",
"verify_evidence", "key_dates", "contract_status", "fetch_executed", "status_of", "executed_bytes"]