Skip to content

fathom.AttestationService

fathom.AttestationService

Signs evaluation results using Ed25519 JWT tokens.

Source code in src/fathom/attestation.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class AttestationService:
    """Signs evaluation results using Ed25519 JWT tokens."""

    def __init__(self, private_key: Ed25519PrivateKey, public_key: Ed25519PublicKey) -> None:
        self._private_key = private_key
        self._public_key = public_key

    @classmethod
    def generate_keypair(cls) -> AttestationService:
        """Generate a new Ed25519 keypair and return an AttestationService."""
        private_key = Ed25519PrivateKey.generate()
        public_key = private_key.public_key()
        return cls(private_key, public_key)

    @classmethod
    def from_private_key_bytes(cls, key_bytes: bytes) -> AttestationService:
        """Create from serialized private key bytes (PEM)."""
        from cryptography.hazmat.primitives.serialization import load_pem_private_key

        private_key = load_pem_private_key(key_bytes, password=None)
        if not isinstance(private_key, Ed25519PrivateKey):
            raise AttestationError("Key is not Ed25519")
        return cls(private_key, private_key.public_key())

    def sign(
        self,
        result: EvaluationResult,
        session_id: str,
        input_facts: list[dict[str, Any]] | None = None,
    ) -> str:
        """Sign an evaluation result and return a JWT token.

        Args:
            result: The evaluation result to attest.
            session_id: Session the evaluation ran under.
            input_facts: The caller-supplied facts the decision was
                computed over. **Required** — the token's ``input_hash``
                is what binds it to a specific fact set, so signing
                without them would issue a token that attests nothing.
                Pass ``[]`` for an evaluation over empty working memory.

        Raises:
            AttestationError: *input_facts* is ``None``.
        """
        if input_facts is None:
            raise AttestationError(
                "sign() requires input_facts: a token signed without them carries "
                "the constant hash of the empty list and binds no inputs"
            )
        # SHA-256 hash of input facts for integrity
        input_hash = hashlib.sha256(json.dumps(input_facts, sort_keys=True).encode()).hexdigest()

        payload = {
            "iss": "fathom",
            "iat": int(time.time()),
            "decision": result.decision,
            "rule_trace": result.rule_trace,
            "input_hash": input_hash,
            "session_id": session_id,
        }

        return self._encode(payload)

    def sign_event(self, payload: dict[str, Any]) -> str:
        """Sign an arbitrary JSON payload and return a JWT token.

        Wraps payload as ``{"iss": "fathom", "iat": <unix ts>, **payload}`` and
        signs with the runtime Ed25519 key. Intended for audit events (e.g.
        hot-reload) that are not shaped like an EvaluationResult.
        """
        claims: dict[str, Any] = {
            "iss": "fathom",
            "iat": int(time.time()),
            **payload,
        }

        return self._encode(claims)

    def sign_claims(self, claims: dict[str, Any], headers: dict[str, Any] | None = None) -> str:
        """Sign a claim set exactly as given (no ``iss``/``iat`` injection).

        Used by :class:`fathom.chained_log.ChainedAttestationLog`, which
        manages its own issuer and timestamps. ``headers`` are added to the
        JWS protected header (e.g. ``kid``).
        """
        return self._encode(claims, headers)

    def _encode(self, claims: dict[str, Any], headers: dict[str, Any] | None = None) -> str:
        try:
            return jwt.encode(claims, self._private_key, algorithm="EdDSA", headers=headers)
        except Exception as exc:
            raise AttestationError(f"Signing failed: {exc}") from exc

    @property
    def public_key(self) -> Ed25519PublicKey:
        return self._public_key

    def public_key_pem(self) -> bytes:
        return self._public_key.public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo)

    def private_key_pem(self) -> bytes:
        return self._private_key.private_bytes(Encoding.PEM, PrivateFormat.PKCS8, NoEncryption())

generate_keypair() classmethod

Generate a new Ed25519 keypair and return an AttestationService.

Source code in src/fathom/attestation.py 
38
39
40
41
42
43
@classmethod
def generate_keypair(cls) -> AttestationService:
    """Generate a new Ed25519 keypair and return an AttestationService."""
    private_key = Ed25519PrivateKey.generate()
    public_key = private_key.public_key()
    return cls(private_key, public_key)

from_private_key_bytes(key_bytes) classmethod

Create from serialized private key bytes (PEM).

Source code in src/fathom/attestation.py 
45
46
47
48
49
50
51
52
53
@classmethod
def from_private_key_bytes(cls, key_bytes: bytes) -> AttestationService:
    """Create from serialized private key bytes (PEM)."""
    from cryptography.hazmat.primitives.serialization import load_pem_private_key

    private_key = load_pem_private_key(key_bytes, password=None)
    if not isinstance(private_key, Ed25519PrivateKey):
        raise AttestationError("Key is not Ed25519")
    return cls(private_key, private_key.public_key())

sign(result, session_id, input_facts=None)

Sign an evaluation result and return a JWT token.

Parameters:

Name Type Description Default
result EvaluationResult

The evaluation result to attest.

 required
session_id str

Session the evaluation ran under.

 required
input_facts list[dict[str, Any]] | None

The caller-supplied facts the decision was computed over. Required — the token's input_hash is what binds it to a specific fact set, so signing without them would issue a token that attests nothing. Pass [] for an evaluation over empty working memory.

 None

Raises:

Type Description
AttestationError

input_facts is None.
Source code in src/fathom/attestation.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def sign(
    self,
    result: EvaluationResult,
    session_id: str,
    input_facts: list[dict[str, Any]] | None = None,
) -> str:
    """Sign an evaluation result and return a JWT token.

    Args:
        result: The evaluation result to attest.
        session_id: Session the evaluation ran under.
        input_facts: The caller-supplied facts the decision was
            computed over. **Required** — the token's ``input_hash``
            is what binds it to a specific fact set, so signing
            without them would issue a token that attests nothing.
            Pass ``[]`` for an evaluation over empty working memory.

    Raises:
        AttestationError: *input_facts* is ``None``.
    """
    if input_facts is None:
        raise AttestationError(
            "sign() requires input_facts: a token signed without them carries "
            "the constant hash of the empty list and binds no inputs"
        )
    # SHA-256 hash of input facts for integrity
    input_hash = hashlib.sha256(json.dumps(input_facts, sort_keys=True).encode()).hexdigest()

    payload = {
        "iss": "fathom",
        "iat": int(time.time()),
        "decision": result.decision,
        "rule_trace": result.rule_trace,
        "input_hash": input_hash,
        "session_id": session_id,
    }

    return self._encode(payload)

sign_event(payload)

Sign an arbitrary JSON payload and return a JWT token.

Wraps payload as {"iss": "fathom", "iat": <unix ts>, **payload} and signs with the runtime Ed25519 key. Intended for audit events (e.g. hot-reload) that are not shaped like an EvaluationResult.

Source code in src/fathom/attestation.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def sign_event(self, payload: dict[str, Any]) -> str:
    """Sign an arbitrary JSON payload and return a JWT token.

    Wraps payload as ``{"iss": "fathom", "iat": <unix ts>, **payload}`` and
    signs with the runtime Ed25519 key. Intended for audit events (e.g.
    hot-reload) that are not shaped like an EvaluationResult.
    """
    claims: dict[str, Any] = {
        "iss": "fathom",
        "iat": int(time.time()),
        **payload,
    }

    return self._encode(claims)

sign_claims(claims, headers=None)

Sign a claim set exactly as given (no iss/iat injection).

Used by :class:fathom.chained_log.ChainedAttestationLog, which manages its own issuer and timestamps. headers are added to the JWS protected header (e.g. kid).

Source code in src/fathom/attestation.py 
109
110
111
112
113
114
115
116
def sign_claims(self, claims: dict[str, Any], headers: dict[str, Any] | None = None) -> str:
    """Sign a claim set exactly as given (no ``iss``/``iat`` injection).

    Used by :class:`fathom.chained_log.ChainedAttestationLog`, which
    manages its own issuer and timestamps. ``headers`` are added to the
    JWS protected header (e.g. ``kid``).
    """
    return self._encode(claims, headers)