117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
 
|
"""Electronic signatures: a step-up check, then an append-only `signatures` row and a `sig.signed` audit row.
|
||
|
|
|
||
|
|
Later phases call `sign` from every sign-off action. There is no HTTP endpoint for it in phase 1.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from datetime import datetime
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from .. import audit
|
||
|
|
from ..settings import Settings, get_settings
|
||
|
|
from . import stepup
|
||
|
|
from .ldap import DirectoryProtocol
|
||
|
|
from .sessions import Actor
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class Signature:
|
||
|
|
id: int
|
||
|
|
at: datetime
|
||
|
|
user_id: Any
|
||
|
|
session_id: Any
|
||
|
|
account: str
|
||
|
|
display_name: str
|
||
|
|
session_kind: str
|
||
|
|
factor: str
|
||
|
|
action: str
|
||
|
|
meaning: str | None
|
||
|
|
packet_id: int | None
|
||
|
|
subject_type: str | None
|
||
|
|
subject_id: str | None
|
||
|
|
content_hash: bytes | None
|
||
|
|
|
||
|
|
|
||
|
|
def canonical_json(content: Any) -> bytes:
|
||
|
|
return json.dumps(content, sort_keys=True, separators=(",", ":"), ensure_ascii=False, default=str).encode("utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
def content_hash(content: Any) -> bytes | None:
|
||
|
|
"""sha256 of the canonical JSON of what was signed; None when nothing was given."""
|
||
|
|
return None if content is None else hashlib.sha256(canonical_json(content)).digest()
|
||
|
|
|
||
|
|
|
||
|
|
def sign(
|
||
|
|
conn: Any,
|
||
|
|
directory: DirectoryProtocol,
|
||
|
|
session: Actor,
|
||
|
|
*,
|
||
|
|
factor: str,
|
||
|
|
secret: str,
|
||
|
|
action: str,
|
||
|
|
meaning: str | None,
|
||
|
|
packet_id: int | None = None,
|
||
|
|
subject_type: str | None = None,
|
||
|
|
subject_id: Any = None,
|
||
|
|
content: Any = None,
|
||
|
|
ip: str | None = None,
|
||
|
|
settings: Settings | None = None,
|
||
|
|
) -> Signature:
|
||
|
|
"""Check the PIN or password (raises ApiError when wrong), then record the signature."""
|
||
|
|
s = settings or get_settings()
|
||
|
|
stepup.verify(conn, directory, s, session, factor=factor, secret=secret, kind="signature", ip=ip)
|
||
|
|
h = content_hash(content)
|
||
|
|
sid = None if subject_id is None else str(subject_id)
|
||
|
|
with conn.transaction():
|
||
|
|
row = conn.execute(
|
||
|
|
"INSERT INTO signatures (user_id, session_id, account, display_name, session_kind, factor, action,"
|
||
|
|
" meaning, packet_id, subject_type, subject_id, content_hash)"
|
||
|
|
" VALUES (%s, %s, %s, %s, %s::session_kind, %s::sig_factor, %s, %s, %s, %s, %s, %s) RETURNING id, at",
|
||
|
|
(
|
||
|
|
session.user_id,
|
||
|
|
session.session_id,
|
||
|
|
session.sam_account,
|
||
|
|
session.display_name,
|
||
|
|
session.kind,
|
||
|
|
factor,
|
||
|
|
action,
|
||
|
|
meaning,
|
||
|
|
packet_id,
|
||
|
|
subject_type,
|
||
|
|
sid,
|
||
|
|
h,
|
||
|
|
),
|
||
|
|
).fetchone()
|
||
|
|
audit.append(
|
||
|
|
conn,
|
||
|
|
action="sig.signed",
|
||
|
|
user_id=session.user_id,
|
||
|
|
session_id=session.session_id,
|
||
|
|
role_snapshot=session.role_snapshot,
|
||
|
|
subject_type=subject_type,
|
||
|
|
subject_id=sid,
|
||
|
|
packet_id=packet_id,
|
||
|
|
detail={"signature_id": int(row[0]), "action": action, "factor": factor},
|
||
|
|
ip=ip,
|
||
|
|
)
|
||
|
|
return Signature(
|
||
|
|
id=int(row[0]),
|
||
|
|
at=row[1],
|
||
|
|
user_id=session.user_id,
|
||
|
|
session_id=session.session_id,
|
||
|
|
account=session.sam_account,
|
||
|
|
display_name=session.display_name,
|
||
|
|
session_kind=session.kind,
|
||
|
|
factor=factor,
|
||
|
|
action=action,
|
||
|
|
meaning=meaning,
|
||
|
|
packet_id=packet_id,
|
||
|
|
subject_type=subject_type,
|
||
|
|
subject_id=sid,
|
||
|
|
content_hash=h,
|
||
|
|
)
|