pamoja.audit

Idiomatic audit-log facade.

A log that can be edited after the fact proves nothing. Each record here is signed and carries the hash of the one before it, so altering a record, dropping one, or reordering two breaks the chain at that point and at every point after it. The index is part of what is signed, which is what makes a record removed from the end detectable too.

 1"""Idiomatic audit-log facade.
 2
 3A log that can be edited after the fact proves nothing. Each record here is
 4signed and carries the hash of the one before it, so altering a record, dropping
 5one, or reordering two breaks the chain at that point and at every point after
 6it. The index is part of what is signed, which is what makes a record removed
 7from the end detectable too.
 8"""
 9
10from __future__ import annotations
11
12from collections.abc import Sequence
13
14from pamoja._native import AuditEntry, AuditVerifier
15from pamoja._native import AuditLog as _AuditLog
16from pamoja._native import verify_audit_chain as _verify_audit_chain
17from pamoja.security import DeviceIdentity
18
19__all__ = [
20    "AuditEntry",
21    "AuditLog",
22    "AuditVerifier",
23    "verify_chain",
24]
25
26
27class AuditLog:
28    """A log that signs what it is given and chains it onto what came before."""
29
30    __slots__ = ("_inner",)
31
32    def __init__(self, identity: DeviceIdentity, last: AuditEntry | None = None) -> None:
33        """Create a log that signs with a device identity.
34
35        :param identity: The identity whose signature each record will carry.
36        :param last: The final record of an existing log to carry on from, or
37            ``None`` to start empty.
38        """
39        signer = DeviceIdentity.native(identity)
40        self._inner = (
41            _AuditLog(signer) if last is None else _AuditLog.resume(signer, last)
42        )
43
44    @staticmethod
45    def resume(identity: DeviceIdentity, last: AuditEntry) -> "AuditLog":
46        """Create a log that carries on from the last record an earlier one wrote.
47
48        This is what a device does after a restart: the chain continues at the
49        next index and hashes onto the record it left off at, so a reboot leaves
50        no gap for a record to be removed through.
51
52        :param identity: The identity whose signature each record will carry.
53        :param last: The final record of the existing log.
54        :returns: The log, positioned after ``last``.
55        """
56        return AuditLog(identity, last)
57
58    def append(self, payload: bytes) -> AuditEntry:
59        """Append a payload, signing it and chaining it onto the last record.
60
61        :param payload: The record to store.
62        :returns: The new entry.
63        """
64        return self._inner.append(payload)
65
66
67def verify_chain(public_key: bytes, entries: Sequence[AuditEntry]) -> None:
68    """Check a whole chain that has already arrived.
69
70    :param public_key: The 32-byte key the records were signed with.
71    :param entries: The records, in the order they were written.
72    :raises PamojaError: With the reason the chain does not hold: a record whose
73        signature fails, or one that does not follow the record before it.
74    """
75    _verify_audit_chain(public_key, list(entries))
class AuditEntry:

One signed record, chained onto the one before it.

def from_bytes(data):

Reads an entry back from the bytes it was written as.

def to_bytes(self, /):

Encodes this entry for storage or transmission.

index

The position of this entry in its chain.

digest

The hash of this entry, which the next one chains onto.

signature

The signature over this entry.

payload

The record this entry carries.

previous

The hash of the entry before this one, all zeroes for the first.

class AuditLog: 
28class AuditLog:
29    """A log that signs what it is given and chains it onto what came before."""
30
31    __slots__ = ("_inner",)
32
33    def __init__(self, identity: DeviceIdentity, last: AuditEntry | None = None) -> None:
34        """Create a log that signs with a device identity.
35
36        :param identity: The identity whose signature each record will carry.
37        :param last: The final record of an existing log to carry on from, or
38            ``None`` to start empty.
39        """
40        signer = DeviceIdentity.native(identity)
41        self._inner = (
42            _AuditLog(signer) if last is None else _AuditLog.resume(signer, last)
43        )
44
45    @staticmethod
46    def resume(identity: DeviceIdentity, last: AuditEntry) -> "AuditLog":
47        """Create a log that carries on from the last record an earlier one wrote.
48
49        This is what a device does after a restart: the chain continues at the
50        next index and hashes onto the record it left off at, so a reboot leaves
51        no gap for a record to be removed through.
52
53        :param identity: The identity whose signature each record will carry.
54        :param last: The final record of the existing log.
55        :returns: The log, positioned after ``last``.
56        """
57        return AuditLog(identity, last)
58
59    def append(self, payload: bytes) -> AuditEntry:
60        """Append a payload, signing it and chaining it onto the last record.
61
62        :param payload: The record to store.
63        :returns: The new entry.
64        """
65        return self._inner.append(payload)

A log that signs what it is given and chains it onto what came before.

AuditLog( identity: pamoja.security.DeviceIdentity, last: AuditEntry | None = None) 
33    def __init__(self, identity: DeviceIdentity, last: AuditEntry | None = None) -> None:
34        """Create a log that signs with a device identity.
35
36        :param identity: The identity whose signature each record will carry.
37        :param last: The final record of an existing log to carry on from, or
38            ``None`` to start empty.
39        """
40        signer = DeviceIdentity.native(identity)
41        self._inner = (
42            _AuditLog(signer) if last is None else _AuditLog.resume(signer, last)
43        )

Create a log that signs with a device identity.

Parameters
  • identity: The identity whose signature each record will carry.
  • last: The final record of an existing log to carry on from, or None to start empty.
@staticmethod
def resume( identity: pamoja.security.DeviceIdentity, last: AuditEntry) -> AuditLog: 
45    @staticmethod
46    def resume(identity: DeviceIdentity, last: AuditEntry) -> "AuditLog":
47        """Create a log that carries on from the last record an earlier one wrote.
48
49        This is what a device does after a restart: the chain continues at the
50        next index and hashes onto the record it left off at, so a reboot leaves
51        no gap for a record to be removed through.
52
53        :param identity: The identity whose signature each record will carry.
54        :param last: The final record of the existing log.
55        :returns: The log, positioned after ``last``.
56        """
57        return AuditLog(identity, last)

Create a log that carries on from the last record an earlier one wrote.

This is what a device does after a restart: the chain continues at the next index and hashes onto the record it left off at, so a reboot leaves no gap for a record to be removed through.

Parameters
  • identity: The identity whose signature each record will carry.
  • last: The final record of the existing log. :returns: The log, positioned after last.
def append(self, payload: bytes) -> AuditEntry: 
59    def append(self, payload: bytes) -> AuditEntry:
60        """Append a payload, signing it and chaining it onto the last record.
61
62        :param payload: The record to store.
63        :returns: The new entry.
64        """
65        return self._inner.append(payload)

Append a payload, signing it and chaining it onto the last record.

Parameters
  • payload: The record to store. :returns: The new entry.
class AuditVerifier:

Checks a chain one entry at a time, in the order the entries were written.

def check(self, /, entry):

Checks the next entry, returning whether it belongs where it was offered.

Feeding entries out of order, skipping one, or repeating one is refused just as an altered payload is.

def verify_chain(public_key: bytes, entries: Sequence[AuditEntry]) -> None: 
68def verify_chain(public_key: bytes, entries: Sequence[AuditEntry]) -> None:
69    """Check a whole chain that has already arrived.
70
71    :param public_key: The 32-byte key the records were signed with.
72    :param entries: The records, in the order they were written.
73    :raises PamojaError: With the reason the chain does not hold: a record whose
74        signature fails, or one that does not follow the record before it.
75    """
76    _verify_audit_chain(public_key, list(entries))

Check a whole chain that has already arrived.

Parameters
  • public_key: The 32-byte key the records were signed with.
  • entries: The records, in the order they were written.
Raises
  • PamojaError: With the reason the chain does not hold: a record whose signature fails, or one that does not follow the record before it.