"""Sending and signing (CT-08 … CT-11, CT-17, CT-52, CT-53).
SEND freezes version N (PDF + sha256), refreshes the counterparty once from business-partner, issues ONE SIGN link
per EXTERNAL signer inside the send's transaction, and records INVITED; each delivery goes through mail.send, or —
with no provider — becomes a manual "copy the signing link" task. Our side (INTERNAL) gets it on their To sign list.
SIGNING: the counterparty through the public door (consent_and_sign, decline, request_changes — public_pages.py calls
them with the door's public identity); our side inside the app as their own sign-in (sign_as_us). Either way: the
version and sha256 shown must be the current frozen ones, it must be the signer's turn, consent must be ticked; the
events are hash-chained; the last signature produces the executed copy in the same request.
"""
from datetime import timedelta
from typing import Literal, Optional
from erp_platform import public, retry
from erp_platform.bricks import Refused
from erp_platform.repository import retry_on_conflict
from erp_platform.types import Name
from brick_contracts.application import support as S
from brick_contracts.application import workflow as W
from brick_contracts.application.use_cases import CT
from brick_contracts.domain.errors import RuleBroken
from brick_contracts.domain.services import Composer
from brick_contracts.domain.values import (EXECUTED, EXTERNAL, INTERNAL, OUT_FOR_SIGNATURE, S_PENDING, SignerRef,
sha256_hex)
from brick_contracts.infrastructure import public_door, renderer
def uow():
from brick_contracts.application import use_cases
return use_cases.repository()
# ── send ───────────────────────────────────────────────────────────────────────────────────
def refresh_counterparty(c, ctx, at) -> list:
"""CT-51: COUNTERPARTY values read through bp once more at send, then frozen in the version. Returns the names
that changed since the draft."""
from brick_contracts.domain.services import BP_FIELDS
from brick_contracts.domain.values import VariableValue
changed = []
for d in ctx.version.variables.values():
if d.source != "COUNTERPARTY":
continue
f = d.validation.get("bp_field") or BP_FIELDS.get(d.name) or d.name
v = ctx.partner.get(f)
if v in (None, ""):
continue
nv = VariableValue.parse(d.name, d.type, v, "COUNTERPARTY", enum_values=d.enum_values, aliases=d.aliases,
resolved_at=S.iso(at))
old = c.variables.get(d.name)
if old is None or old.value != nv.value:
c.variables[d.name] = nv
changed.append(d.name)
return changed
def send(caller: dict, contract_id: str, *, check) -> tuple:
"""(Contract, deliveries, replayed). `check(view, c)` raises 404 unless the caller may send this contract."""
with uow() as u:
view = S.View(u, caller)
h = u.contracts.head(contract_id) # one SELECT: may they, and is it already out?
if h is None:
raise S.not_found(f"contract {contract_id}")
check(view, h)
if h.state == OUT_FOR_SIGNATURE or h.state in EXECUTED:
c = u.contracts.get(contract_id)
return c, [], True
bp_id = h.bp_id
partner = S.partner(bp_id, S.actor(caller))
urls: dict = {}
def go():
urls.clear()
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
cfg = view.config
c = u.contracts.get(contract_id)
check(view, c)
if c.state == OUT_FOR_SIGNATURE or c.state in EXECUTED:
return c, True
c.require_editable("send it")
ctx = S.drafting_for(u, cfg, c, partner_row=partner)
if refresh_counterparty(c, ctx, at):
from brick_contracts.application.drafting import review
review(c, ctx, at=at, by=S.actor(caller))
refusals = c.send_refusals(ctx.ctype, S.pack_refusals(cfg, c))
if refusals:
S.settle(c, cfg, at)
u.contracts.save(c, now=at) # the refreshed values and any approval it opened stay
return c, "; ".join(refusals)
n = max(c.versions, default=0) + 1
text = Composer.compose(title=c.title, our_name=cfg.setting("our_legal_name", "") or "Us",
counterparty_name=partner.get("legal_name") or "", version=n,
contract_id=c.contract_id, sections=c.sections, variables=c.variables,
signers=c.signers)
pdf = renderer.render(text, footer=f"{c.contract_id} version {n}")
key = f"contracts/{c.contract_id}/v{n}.pdf"
sha, size = u.objects.put(key, pdf, "application/pdf")
c.freeze(object_key=key, sha256=sha, size_bytes=size, text_body=text, now=at, by=S.actor(caller),
link_lifetime=timedelta(days=S.link_days(cfg)))
for s in c.signers:
if s.kind == EXTERNAL:
issued = public_door.issue(c.contract_id, s.signer_no, s.email, "SIGN", days=S.link_days(cfg),
actor=S.actor(caller), conn=u.conn)
s.link_id = issued.link_id
subj, body = W.invite(c, s, cfg, None)
d = W.new_delivery(c, s.signer_no, "INVITE", s.email, subj, body, at)
urls[(s.signer_no, "INVITE", d.seq)] = issued.url
S.settle(c, cfg, at)
u.contracts.save(c, now=at)
return c, False
c, replayed = retry_on_conflict(go)
if isinstance(replayed, str):
raise _NotSent(replayed) # after the commit: what the attempt learned is kept
if not replayed:
W.deliver(c.contract_id, [(k[0], k[1], k[2], v) for k, v in urls.items()])
with uow() as u:
c = u.contracts.get(c.contract_id)
return c, [d for d in c.deliveries if d.kind == "INVITE"], replayed
class _NotSent(Exception):
"""Send refused, in words (invariant 4). The doors answer 409; the capability answers failed."""
def send_answer(c) -> dict:
v = c.current
return {"contract_id": c.contract_id, "state": c.state, "version": c.current_version, "sha256": v.sha256 if v else "",
"deliveries": [{"signer_no": d.signer_no, "channel": d.channel} for d in c.deliveries
if d.kind == "INVITE" and v is not None and d.created_at and d.created_at >= v.frozen_at]}
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/send", permission=S.SEND, idempotent=True)
def send_for_signature(caller: dict, contract_id: S.ContractId) -> dict:
"""Send for signature in one step (CT-08): version N is rendered to PDF and frozen with its sha256; each EXTERNAL
signer gets one signing link (mail.send, or with no mail provider a manual 'copy the signing link' task); our
INTERNAL signers find it on their To sign list. Refused, in words, while values are missing, an approval is open,
the type is a deed or not signed online, only one side is named, or a UK check fails (an outside IR35
determination by a medium or large client without its status determination statement). Sending twice is one
version."""
def check(view, c):
if not view.manages(c):
raise S.not_found(f"contract {contract_id}")
try:
c, _, _ = send(caller, contract_id, check=check)
except _NotSent as e:
raise Refused(409, str(e)) from None
return send_answer(c)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/recall", permission=S.RECALL, idempotent=True)
def recall(caller: dict, contract_id: S.ContractId, reason: S.Reason) -> dict:
"""Recall a contract out for signature: every signing link stops working (REVOKED), the reason is recorded, and it
is a DRAFT again — the next send is version N+1 (a signer still looking at version N is refused). Recalling one
that is not out is nothing."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = view.contract(contract_id, manage=True)
if c.state != OUT_FOR_SIGNATURE:
return S.full(c, view)
for lid in c.recall(reason, now=at):
public_door.revoke(lid, actor=S.actor(caller), reason="recalled", conn=u.conn)
S.settle(c, view.config, at)
u.contracts.save(c, now=at)
return S.full(c, view)
return retry_on_conflict(go)
def _reissue(caller: dict, contract_id: str, signer_no: int, which: str) -> tuple:
"""Revoke the signer's live link and issue its replacement (REISSUED in the chain); the url once."""
holder = {}
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = view.contract(contract_id, manage=True)
s = c.signer(signer_no)
if s.kind != EXTERNAL:
raise RuleBroken(f"{s.full_name} signs inside the app; there is no link to copy")
cfg = view.config
if which == "sign":
if c.state != OUT_FOR_SIGNATURE or s.status != S_PENDING:
raise RuleBroken("there is no signing link to copy: the contract is not waiting for this signer")
old = s.link_id
issued = (public_door.reissue(old, actor=S.actor(caller), conn=u.conn) if old else
public_door.issue(c.contract_id, s.signer_no, s.email, "SIGN", days=S.link_days(cfg),
actor=S.actor(caller), conn=u.conn))
s.link_id = issued.link_id
else:
if c.executed is None:
raise RuleBroken("there is no signed copy yet")
old = s.copy_link_id
issued = (public_door.reissue(old, actor=S.actor(caller), conn=u.conn) if old else
public_door.issue(c.contract_id, s.signer_no, s.email, "READ", days=S.copy_link_days(cfg),
actor=S.actor(caller), conn=u.conn))
s.copy_link_id = issued.link_id
c.append_event("REISSUED", at, s, version=c.current_version, token_ref=issued.link_id,
detail=f"{which} link replaced{(' ' + old) if old else ''} by {S.actor(caller)}")
for d in c.deliveries:
if d.signer_no == s.signer_no and d.status == "pending" and d.channel == "manual" and (
(which == "sign" and d.kind in ("INVITE", "REISSUE") or d.kind.startswith("CHASE"))
or (which == "copy" and d.kind == "COPY")):
d.plan = f"link copied by {S.actor(caller)} at {S.iso(at)}"
S.settle(c, cfg, at)
u.contracts.save(c, now=at)
holder["c"] = c
return issued
issued = retry_on_conflict(go)
return holder["c"], issued
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/signers/{signer_no}/copy-link", permission=S.SEND)
def copy_signing_link(caller: dict, contract_id: S.ContractId, signer_no: S.SignerNo,
which: Literal["sign", "copy"] = "sign", idempotency_key: Optional[str] = None) -> dict:
"""Copy a signer's link to send it yourself (CT-52) — the manual task when there is no mail provider: a fresh link
is issued, the previous one stops working (REISSUED is written into the evidence), and the url is in THIS answer
only — it is never stored or logged. which = sign (the signing link) or copy (the signed-copy link). Mark the task
delivered once you have sent it. (idempotency_key is not honoured here on purpose: replaying this answer would mean
storing the link.)"""
c, issued = _reissue(caller, contract_id, signer_no, which)
return {"contract_id": c.contract_id, "signer_no": int(signer_no), "which": which, "url": issued.url,
"note": "shown once — copy it now; asking again issues a new link and this one stops working"}
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/signers/{signer_no}/reissue", permission=S.SEND)
def reissue_link(caller: dict, contract_id: S.ContractId, signer_no: S.SignerNo, idempotency_key: Optional[str] = None) -> dict:
"""Reissue a signer's signing link (CT-11): the old one stops working (REISSUED) and the new one goes out through
mail.send; with no mail provider the new url is in this answer, once. (idempotency_key is not honoured: the
answer may carry the link.)"""
from brick_contracts.infrastructure import mail
c, issued = _reissue(caller, contract_id, signer_no, "sign")
if mail.provider():
def add():
at = S.now()
with uow() as u:
c2 = u.contracts.get(contract_id)
s = c2.signer(signer_no)
subj, body = W.invite(c2, s, u.config.load(), None)
d = W.new_delivery(c2, s.signer_no, "REISSUE", s.email, subj, body, at)
u.contracts.save(c2, now=at)
return d.seq
seq = retry_on_conflict(add)
out = W.deliver(contract_id, [(int(signer_no), "REISSUE", seq, issued.url)])
return {"contract_id": contract_id, "signer_no": int(signer_no), "delivered": out}
return {"contract_id": contract_id, "signer_no": int(signer_no), "url": issued.url,
"note": "no mail provider here: send this link yourself — it is shown once"}
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/deliveries/delivered", permission=S.SEND,
idempotent=True)
def mark_delivered_by_hand(caller: dict, contract_id: S.ContractId, signer_no: int, kind: S.Code, seq: S.SeqNo) -> dict:
"""Mark a manual delivery done — you sent the link, the reminder or the copy yourself (CT-52). For a signing link
it writes DELIVERED (manual) into the evidence. Twice is once."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c = view.contract(contract_id, manage=True)
d = c.delivery(signer_no, kind.upper() if ":" not in kind else kind, seq)
if d.status == "pending":
d.status, d.done_by, d.done_at = "done", S.actor(caller), at
if d.signer_no and (d.kind in ("INVITE", "REISSUE") or d.kind.startswith("CHASE")):
c.mark_delivered(d.signer_no, now=at, by=S.actor(caller), channel="manual", provider_ref=None)
S.settle(c, view.config, at)
u.contracts.save(c, now=at)
return S.delivery_json(d)
return retry_on_conflict(go)
# ── our side, inside the app (CT-17) ───────────────────────────────────────────────────────
def _mine(view, contract_id: str):
c = view.uow.contracts.get(contract_id)
if c is None:
raise S.not_found(f"contract {contract_id}")
s = next((x for x in c.signers if x.kind == INTERNAL and x.is_me(view.caller)), None)
if s is None:
raise S.not_found(f"contract {contract_id} to sign") # not this signer: 404, never 403
return c, s
def _signing_view(c, s, cfg) -> dict:
v = c.current
turn = c.whose_turn()
consent = cfg.consent()
return {"contract_id": c.contract_id, "title": c.title, "signer_no": s.signer_no, "status": s.status,
"version": v.version if v else None, "sha256": v.sha256 if v else None,
"text": v.text_body if v else None, "my_turn": s in turn,
"waiting_on": [x.full_name for x in turn if x is not s],
"consent_text": consent.text if consent else None, "consent_text_version": consent.version if consent else None,
"retention": S.retention_note(cfg, c), "claim": "signed electronically"}
@CT.use_case(kind="query", route="GET /to-sign", permission=S.SIGN)
def to_sign(caller: dict) -> dict:
"""Your To sign list (CT-17): the contracts out for signature where you sign for our side, with the frozen version,
its fingerprint, the consent text and whether it is your turn."""
with uow() as u:
view = S.View(u, caller)
items = []
for c in u.contracts.load_many(view.signing):
if c.state != OUT_FOR_SIGNATURE:
continue
s = next((x for x in c.signers if x.kind == INTERNAL and x.is_me(caller)), None)
if s is not None and s.status == S_PENDING:
items.append(_signing_view(c, s, view.config))
return {"to_sign": items}
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/open-as-us", permission=S.SIGN)
def open_as_us(caller: dict, contract_id: S.ContractId) -> dict:
"""Open a contract you sign for our side: records OPENED (INTERNAL, POOL_SESSION, your pool sub) and returns the
frozen version's words and sha256 — what you sign."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c, s = _mine(view, contract_id)
if c.state != OUT_FOR_SIGNATURE:
raise RuleBroken("this contract is not out for signature")
c.record_opened(s.signer_no, now=at, via=INTERNAL, pool_sub=caller.get("sub"),
session_id=caller.get("session_id"))
u.contracts.save(c, now=at)
return _signing_view(c, s, view.config)
return retry_on_conflict(go)
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/sign", permission=S.SIGN, idempotent=True)
def sign_as_us(caller: dict, contract_id: S.ContractId, version: S.VersionNo, sha256: S.Code, typed_name: Name,
consent: bool) -> dict:
"""Sign for our side inside the app, as yourself (CT-17): you send the version and sha256 you were shown, tick the
consent and type your name. CONSENTED and SIGNED are written with signer kind INTERNAL, auth_method POOL_SESSION
and your pool sub. Refused when it is not your turn (naming who it waits on) or the version was re-issued; a
caller who is not this signer gets 404. Signing twice is one signature. The last signature produces the executed
copy at once."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c, s = _mine(view, contract_id)
consent_text = view.config.consent()
if consent_text is None:
raise RuleBroken("no consent text is set up — keep one in settings first")
did = c.sign(s.signer_no, version=version, sha_shown=sha256.lower(), typed_name=typed_name,
consented=consent, consent=consent_text, now=at, via=INTERNAL, pool_sub=caller.get("sub"),
session_id=caller.get("session_id"))
if did:
S.settle(c, view.config, at)
u.contracts.save(c, now=at)
return c, s
c, s = retry_on_conflict(go)
executed = c.executed is not None
if c.all_signed and not executed:
executed = bool((W.finalize(c.contract_id) or {}).get("executed"))
return {"contract_id": c.contract_id, "signer_no": s.signer_no, "status": s.status,
"state": "SIGNED" if executed else c.state, "signed_version": s.signed_version, "executed": executed}
@CT.use_case(kind="command", route="POST /contracts/{contract_id}/decline-as-us", permission=S.SIGN, idempotent=True)
def decline_as_us(caller: dict, contract_id: S.ContractId, version: S.VersionNo, sha256: S.Code, reason: S.Reason) -> dict:
"""Decline to sign for our side, with a reason: the contract is DECLINED and every link revoked."""
def go():
at = S.now()
with S.rules(), uow() as u:
view = S.View(u, caller, at)
c, s = _mine(view, contract_id)
if s.status == "DECLINED":
return {"contract_id": c.contract_id, "state": c.state}
for lid in c.decline(s.signer_no, reason, now=at, via=INTERNAL, version=version, sha_shown=sha256.lower(),
pool_sub=caller.get("sub"), session_id=caller.get("session_id")):
public_door.revoke(lid, actor=S.actor(caller), reason="declined", conn=u.conn)
W.new_delivery(c, 0, "NOTICE:declined", c.owner or c.owner_group, f"Declined: {c.title}", c.last_note, at)
S.settle(c, view.config, at)
u.contracts.save(c, now=at)
return {"contract_id": c.contract_id, "state": c.state}
return retry_on_conflict(go)
# ── the counterparty, through the public door (CT-09, CT-10, CT-53) ────────────────────────
def _subject(caller: dict, subject_id: str, grant: str) -> SignerRef:
"""The public caller's own subject, or 404: a link reaches only its own signer's page."""
sub = public.subject_of(caller)
if sub is None or sub[0] != public_door.SUBJECT or sub[1] != subject_id or sub[2] != grant:
raise Refused(404, "not found")
return SignerRef.parse(subject_id)
def open_for_signing(caller: dict, subject_id: str, *, ip=None, user_agent=None) -> dict:
"""GET of a signing link: OPENED with IP and user agent as fields; the frozen version to read."""
ref = _subject(caller, subject_id, "SIGN")
link_id = caller["public_link"]["link_id"]
def go():
at = S.now()
with S.rules(), uow() as u:
c = u.contracts.get(ref.contract_id)
if c is None or c.state != OUT_FOR_SIGNATURE:
raise Refused(404, "not found")
s = c.signer(ref.signer_no)
if s.kind != EXTERNAL or s.link_id != link_id:
raise Refused(404, "not found")
cfg = u.config.load()
c.record_opened(s.signer_no, now=at, via=EXTERNAL, ip=ip, user_agent=user_agent, token_ref=link_id)
S.settle(c, cfg, at)
u.contracts.save(c, now=at)
return _signing_view(c, s, cfg) | {"owner": c.owner or c.owner_group, "our_name": W.our_name(cfg)}
return retry_on_conflict(go)
def frozen_pdf(caller: dict, subject_id: str) -> tuple:
"""The frozen bytes of the version out for signature (READ on a SIGN link) — served as stored, never re-rendered."""
ref = _subject(caller, subject_id, "SIGN")
with uow() as u:
c = u.contracts.get(ref.contract_id)
if c is None or c.state != OUT_FOR_SIGNATURE or c.signer(ref.signer_no).link_id != caller["public_link"]["link_id"]:
raise Refused(404, "not found")
got = u.objects.get(c.current.object_key)
return got[0], f"{c.contract_id}-v{c.current_version}.pdf"
def _act(caller: dict, subject_id: str, link, act) -> object:
"""A signer's final act, with the link's single-use claim: inside our transaction when the link tables share our
engine (link.use(conn)), else claimed just before and released if the act is refused."""
ref = _subject(caller, subject_id, "SIGN")
link_id = caller["public_link"]["link_id"]
claimed_outside = {"done": False}
def go():
at = S.now()
with S.rules(), uow() as u:
c = u.contracts.get(ref.contract_id)
if c is None:
raise Refused(404, "not found")
s = c.signer(ref.signer_no)
if s.kind != EXTERNAL or s.link_id != link_id:
raise Refused(404, "not found")
cfg = u.config.load()
out = act(c, s, cfg, at, u)
shared = public_door.shared(u.conn)
if shared is not None:
link.use(shared) # claimed in OUR transaction
elif not claimed_outside["done"]:
if not link.use():
raise Refused(404, "not found")
claimed_outside["done"] = True
S.settle(c, cfg, at)
u.contracts.save(c, now=at)
return c, out
try:
return retry.run(lambda: retry_on_conflict(go), idempotent=False)
except Exception:
if claimed_outside["done"]:
link.release("the act did not complete")
raise
def consent_and_sign(caller: dict, subject_id: str, *, typed_name: str, consented: bool, version: int, sha256: str,
ip=None, user_agent=None, link=None) -> dict:
def act(c, s, cfg, at, u):
consent = cfg.consent()
if consent is None:
raise RuleBroken("signing is not set up yet (no consent text) — ask the sender")
if not c.sign(s.signer_no, version=int(version), sha_shown=(sha256 or "").lower(), typed_name=typed_name,
consented=consented, consent=consent, now=at, via=EXTERNAL, ip=ip, user_agent=user_agent,
token_ref=s.link_id):
raise Refused(404, "not found") # already signed: the link is spent
return None
c, _ = _act(caller, subject_id, link, act)
if c.all_signed:
W.finalize(c.contract_id)
return {"contract_id": c.contract_id, "signed": True}
def decline(caller: dict, subject_id: str, *, reason: str, version: int, sha256: str, ip=None, user_agent=None,
link=None) -> dict:
def act(c, s, cfg, at, u):
for lid in c.decline(s.signer_no, reason, now=at, via=EXTERNAL, version=int(version),
sha_shown=(sha256 or "").lower(), ip=ip, user_agent=user_agent, token_ref=s.link_id):
if lid != s.link_id:
public_door.revoke(lid, actor="signer", reason="declined", conn=u.conn)
W.new_delivery(c, 0, "NOTICE:declined", c.owner or c.owner_group, f"Declined: {c.title}", c.last_note, at)
_act(caller, subject_id, link, act)
return {"declined": True}
def request_changes(caller: dict, subject_id: str, *, note: str, version: int, sha256: str, ip=None, user_agent=None,
link=None) -> dict:
def act(c, s, cfg, at, u):
for lid in c.request_changes(s.signer_no, note, now=at, via=EXTERNAL, version=int(version),
sha_shown=(sha256 or "").lower(), ip=ip, user_agent=user_agent, token_ref=s.link_id):
if lid != s.link_id:
public_door.revoke(lid, actor="signer", reason="changes requested", conn=u.conn)
W.new_delivery(c, 0, "NOTICE:changes", c.owner or c.owner_group, f"Changes asked for: {c.title}", c.last_note,
at)
_act(caller, subject_id, link, act)
return {"changes_requested": True}
def signed_copy(caller: dict, subject_id: str) -> tuple:
"""READ link: the executed copy's bytes, checked against their sha256 and the evidence chain first."""
from brick_contracts.domain.services import ChainVerifier
ref = _subject(caller, subject_id, "READ")
with uow() as u:
c = u.contracts.get(ref.contract_id)
if c is None or c.executed is None or c.signer(ref.signer_no).copy_link_id != caller["public_link"]["link_id"]:
raise Refused(404, "not found")
ok, _ = ChainVerifier.verify(c.contract_id, c.events)
got = u.objects.get(c.executed.object_key)
if not ok or got is None or sha256_hex(got[0]) != c.executed.sha256:
raise Refused(409, "this signed copy cannot be released: its evidence does not verify — ask the sender")
return got[0], c.executed.filename
__all__ = ["send", "send_answer", "send_for_signature", "recall", "copy_signing_link", "reissue_link",
"mark_delivered_by_hand", "to_sign", "open_as_us", "sign_as_us", "decline_as_us", "open_for_signing",
"frozen_pdf", "consent_and_sign", "decline", "request_changes", "signed_copy", "_NotSent"]