Skip to content

Provenance

ProvenanceTagger

ProvenanceTagger 

ProvenanceTagger(source_id: str, signing_key: bytes | None = None, include_device_attestation: bool = False, agent_version: str = '0.1.0')

Tags events with provenance metadata for trust verification.

Provenance includes: - Capture timestamp (from source clock) - Send timestamp (when leaving SDK) - Sequence number (monotonic) - Source signature (if signing key provided) - SDK version - Agent ID (unique per agent instance)

Initialize the provenance tagger.

PARAMETER DESCRIPTION
source_id

Identifier of the source system

TYPE: str

signing_key

Optional key for signing events (HMAC-SHA256)

TYPE: bytes | None DEFAULT: None

include_device_attestation

Whether to include device attestation

TYPE: bool DEFAULT: False

agent_version

SDK/agent version string

TYPE: str DEFAULT: '0.1.0'

Source code in phytrace/core/provenance.py 
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
def __init__(
    self,
    source_id: str,
    signing_key: bytes | None = None,
    include_device_attestation: bool = False,
    agent_version: str = "0.1.0",
):
    """
    Initialize the provenance tagger.

    Args:
        source_id: Identifier of the source system
        signing_key: Optional key for signing events (HMAC-SHA256)
        include_device_attestation: Whether to include device attestation
        agent_version: SDK/agent version string
    """
    self.source_id = source_id
    self.signing_key = signing_key
    self.include_device_attestation = include_device_attestation
    self.agent_version = agent_version

    # Generate a unique agent instance ID
    self.agent_id = f"agent-{uuid.uuid4().hex[:12]}"

    # Monotonic sequence counter
    self._sequence_num = 0

    # Track last capture time for clock skew detection
    self._last_capture_time: datetime | None = None

Attributes

current_sequence property 

current_sequence: int

Get the current sequence number.

Functions

tag 

tag(event: UDMEvent) -> TaggedEvent

Add provenance metadata to event.

PARAMETER DESCRIPTION
event

The UDM event to tag

TYPE: UDMEvent

RETURNS DESCRIPTION
TaggedEvent

TaggedEvent with provenance metadata
Source code in phytrace/core/provenance.py 
 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
133
134
135
def tag(self, event: UDMEvent) -> TaggedEvent:
    """
    Add provenance metadata to event.

    Args:
        event: The UDM event to tag

    Returns:
        TaggedEvent with provenance metadata
    """
    # Increment sequence number
    self._sequence_num += 1

    # Get timestamps
    captured_at = event.captured_at
    sent_at = datetime.utcnow()

    # Detect clock skew
    clock_skew_detected = False
    if self._last_capture_time and captured_at < self._last_capture_time:
        clock_skew_detected = True
    self._last_capture_time = captured_at

    # Update event with send timestamp and sequence
    event.sent_at = sent_at
    event.sequence_num = self._sequence_num

    # Generate signature if signing key provided
    signature = None
    signature_algorithm = None
    if self.signing_key:
        signature = self._sign_event(event)
        signature_algorithm = "HMAC-SHA256"

    # Get device attestation if enabled
    device_attestation = None
    if self.include_device_attestation:
        device_attestation = self._get_device_attestation()

    # Create provenance metadata
    provenance = ProvenanceMetadata(
        agent_id=self.agent_id,
        agent_version=self.agent_version,
        sequence_num=self._sequence_num,
        captured_at=captured_at,
        sent_at=sent_at,
        clock_skew_detected=clock_skew_detected,
        signature=signature,
        signature_algorithm=signature_algorithm,
        device_attestation=device_attestation,
    )

    return TaggedEvent(event=event, provenance=provenance)

reset_sequence 

reset_sequence() -> None

Reset the sequence counter (useful for new sessions).

Source code in phytrace/core/provenance.py 
173
174
175
176
def reset_sequence(self) -> None:
    """Reset the sequence counter (useful for new sessions)."""
    self._sequence_num = 0
    self._last_capture_time = None

ProvenanceMetadata

ProvenanceMetadata dataclass 

ProvenanceMetadata(agent_id: str, agent_version: str, sequence_num: int, captured_at: datetime, sent_at: datetime, clock_skew_detected: bool = False, signature: str | None = None, signature_algorithm: str | None = None, device_attestation: dict[str, Any] | None = None)

Provenance metadata added to events.