crp.security¶

Auto-generated reference for the crp.security subpackage.

`security`¶

`crp.security` ¶

`ComplianceAuditTrail` ¶

Tamper-evident, HMAC-signed compliance audit trail (§7.14).

Provides an immutable, append-only log with cryptographic chaining that detects any modification to historical entries.

EU AI Act Art. 12: "High-risk AI systems shall technically allow for the automatic recording of events (logs) over the lifetime of the system."

ISO 42001 A.6.2.8: Organizations must maintain records of AI system development, deployment, operation, monitoring, and decommissioning.

Usage::

trail = ComplianceAuditTrail(signing_key=session_key)
trail.record(
    ComplianceEventType.DATA_INGESTED,
    session_id="abc-123",
    data={"source": "user-input", "size_bytes": 4096},
)
# Verify chain integrity
valid, broken_at = trail.verify_chain()
assert valid
# Export for regulatory review
export = trail.export()

`entry_count` `property` ¶

Number of entries currently stored in the trail.

`record(event_type, session_id='', data=None)` ¶

Append a tamper-evident entry to the audit trail.

Thread-safe. Each entry is chained to the previous via HMAC.

Parameters:

Name	Type	Description	Default
`event_type`	`ComplianceEventType \| str`	Compliance event type enum or custom string.	required
`session_id`	`str`	Session identifier. Falls back to the trail default.	`''`
`data`	`dict[str, Any] \| None`	Arbitrary event payload.	`None`

Returns:

Type	Description
`AuditEntry`	The newly created `AuditEntry`.

`verify_chain()` ¶

Verify the integrity of the entire audit trail.

Returns:

Type	Description
`bool`	`(is_valid, broken_at_sequence)`. `broken_at_sequence` is `-1`
`int`	when the chain is valid, otherwise the sequence of the first broken entry.

`query(*, event_type=None, session_id=None, since=None, until=None)` ¶

Query audit entries with optional filters.

Parameters:

Name	Type	Description	Default
`event_type`	`str \| ComplianceEventType \| None`	Filter by event type.	`None`
`session_id`	`str \| None`	Filter by session ID.	`None`
`since`	`float \| None`	Include entries with timestamp >= this value.	`None`
`until`	`float \| None`	Include entries with timestamp <= this value.	`None`

Returns:

Type	Description
`list[AuditEntry]`	Matching `AuditEntry` objects in chronological order.

`export(*, include_signatures=True, since=None)` ¶

Export the full audit trail for regulatory review.

Parameters:

Name	Type	Description	Default
`include_signatures`	`bool`	Whether to include HMAC signatures in the export.	`True`
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`dict[str, Any]`	Structured document suitable for compliance auditors, including
`dict[str, Any]`	chain integrity status and summary statistics.

`export_jsonl(*, since=None)` ¶

Export as JSONL (one entry per line) for log aggregation.

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`str`	Newline-delimited JSON string of all matching entries.

`export_ndjson(*, since=None)` ¶

Alias for :meth:export_jsonl - NDJSON is the primary format (§4.1).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`str`	Newline-delimited JSON string of all matching entries.

`export_ocsf(*, since=None, provider='', model='')` ¶

Export events in OCSF API Activity format for SIEMs (§4.2).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`
`provider`	`str`	Provider identifier for the destination endpoint field.	`''`
`model`	`str`	Model identifier for the destination endpoint field.	`''`

Returns:

Type	Description
`list[dict[str, Any]]`	List of OCSF `class_uid=6003` records. The HMAC and risk level
`list[dict[str, Any]]`	are carried in the `unmapped` extension.

`export_sarif(*, since=None)` ¶

Export events as a SARIF 2.1.0 log for GitHub integration (§4.3).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`dict[str, Any]`	SARIF 2.1.0 log dict. Non-INFO events become SARIF results; INFO
`dict[str, Any]`	events are emitted at `note` level. Used by CRP Scan (CRP-SPEC-013).

`ComplianceEventType` ¶

Bases: str, Enum

Compliance-grade event types beyond standard CRP events (§7.14.1).

`SessionBindingManager` ¶

HMAC-SHA256 session binding with OS keyring + zero-config fallback (§7.1).

Usage

mgr = SessionBindingManager() binding = mgr.create_session("session-123") sig = mgr.sign_request(b"payload") assert mgr.verify_request_signature(b"payload", sig)

`binding` `property` ¶

Return the binding.

`session_key` `property` ¶

Return current session key (raises if no session).

`key_version` `property` ¶

Current key version number.

`create_session(session_id='')` ¶

Create a new session with fresh nonce and derived key (§6A.1, §6A.2).

session_nonce: 32 bytes of cryptographic randomness
session_key: HMAC-SHA256(master_secret, nonce || session_id)

`sign_request(payload)` ¶

Compute HMAC-SHA256 signature over payload (§6A.3).

`verify_request_signature(payload, signature)` ¶

Constant-time HMAC verification (§6A.3).

Uses hmac.compare_digest to prevent timing attacks. Also checks previous key versions during rotation window.

`rotate_secret()` ¶

Rotate master secret with graceful rollover (§audit H2).

Generates a fresh 256-bit master secret, re-derives the session key, and keeps the previous key(s) valid for verification during the rotation window.

Returns:

Type	Description
`SessionBinding`	Updated SessionBinding with the new key material.

Raises:

Type	Description
`RuntimeError`	If no active session exists.

`store_to_keyring(service_name='crp-sdk')` ¶

Store master secret in OS keyring (DPAPI/Keychain/kernel).

Returns True if stored, False if keyring unavailable.

`from_keyring(service_name='crp-sdk')` `classmethod` ¶

Load master secret from OS keyring. Returns None if unavailable.

`Checkpoint` `dataclass` ¶

Inline human-in-the-loop declaration (SPEC-033 §3).

Attributes:

Name	Type	Description
`checkpoint_id`	`str`	Unique identifier for this checkpoint instance.
`trigger`	`CheckpointTrigger`	Condition that caused the checkpoint to fire.
`timeout`	`int`	Seconds before auto-action is taken.
`on_timeout`	`CheckpointTimeoutAction`	Action if human does not respond in time.
`on_reject`	`CheckpointRejectAction`	Action if human rejects.
`context`	`dict[str, Any]`	Arbitrary context dict for the reviewer UI.

`wait_for_resolution()` `async` ¶

Block until the checkpoint is resolved by a human reviewer.

If timeout expires, auto-resolves per on_timeout.

Returns:

Type	Description
`CheckpointResolution`	A `CheckpointResolution` - never `None` (Invariant 10).

Raises:

Type	Description
`TimeoutError`	Internally caught and converted to auto-resolution.

`resolve(resolution)` ¶

Called by a human reviewer (or webhook) to resolve this checkpoint.

Parameters:

Name	Type	Description	Default
`resolution`	`CheckpointResolution`	The reviewer's decision and optional edited output.	required

Returns:

Type	Description
`None`	None. Sets the internal resolved event so awaiting tasks unblock.

`decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback')` `classmethod` ¶

Decorator factory: @checkpoint(when="risk >= HIGH").

Parameters:

Name	Type	Description	Default
`when`	`str`	Trigger condition string.	`'always'`
`timeout`	`int`	Seconds to await human review.	`300`
`on_timeout`	`str`	Action if the checkpoint times out.	`'escalate'`
`on_reject`	`str`	Action if the reviewer rejects.	`'fallback'`

Returns:

Type	Description
`Callable`	A decorator that wraps the function with checkpoint gating.

Raises:

Type	Description
`CRPError`	If the reviewer rejects and `on_reject` is `"halt"`.

`CheckpointRejectAction` ¶

Bases: str, Enum

What happens when a checkpoint is rejected by the human reviewer (SPEC-033 §3).

`CheckpointResolution` `dataclass` ¶

The result of a human reviewer resolving a checkpoint (SPEC-033 §3).

Attributes:

Name	Type	Description
`action`	`CheckpointResolutionAction`	Reviewer decision (approve/reject/edit).
`reviewer`	`str`	Identifier of the reviewer.
`timestamp`	`float`	Unix timestamp of the resolution.
`edited_output`	`str`	Optional replacement text when action is EDIT.
`audit_event`	`dict[str, Any] \| None`	Optional extra fields for the audit trail.

`to_audit_dict()` ¶

Produce an audit-trail compatible event dict (SPEC-011).

Returns:

Type	Description
`dict[str, Any]`	Dict with checkpoint resolution fields ready for `ComplianceAuditTrail`.

`CheckpointResolutionAction` ¶

Bases: str, Enum

The human reviewer's decision (SPEC-033 §3).

`CheckpointTimeoutAction` ¶

Bases: str, Enum

What happens when a checkpoint times out awaiting human response (SPEC-033 §3).

`CheckpointTrigger` ¶

Bases: str, Enum

What causes a checkpoint to fire (SPEC-033 §3).

`ClarificationAction` ¶

Bases: str, Enum

The user's response to a clarification request.

`ClarificationRequest` `dataclass` ¶

A request for user input raised by a CLARIFY operation or an oversight gate.

`to_dict()` ¶

Render for streaming / audit / a reviewer UI.

`ClarificationResolution` `dataclass` ¶

The outcome of resolving a clarification request.

`approved` `property` ¶

True when the resolution is an affirmative answer (for oversight gates).

`AIRiskLevel` ¶

Bases: str, Enum

EU AI Act risk classification levels (Art. 6) (§7.15.1).

`AISystemCategory` ¶

Bases: str, Enum

Categories of AI system use cases relevant to risk classification.

`ComplianceReporter` ¶

Generate compliance status reports (§7.15.3).

Maps CRP's native security controls to EU AI Act articles and ISO 42001 clauses, reporting implementation status for each.

Usage::

reporter = ComplianceReporter()
report = reporter.generate_report(session_stats={...})
print(report["summary"]["compliance_score"])

`generate_report(session_stats=None, risk_assessment=None)` ¶

Generate a comprehensive compliance status report.

Parameters:

Name	Type	Description	Default
`session_stats`	`dict[str, Any] \| None`	Optional runtime/session statistics to include.	`None`
`risk_assessment`	`RiskAssessment \| None`	Optional `RiskAssessment` to attach.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with EU AI Act and ISO 42001 control lists, implementation
`dict[str, Any]`	counts, compliance percentages, and summary score.

`generate_technical_documentation(transparency=None, risk_assessment=None)` ¶

Generate EU AI Act Art. 11 technical documentation.

Parameters:

Name	Type	Description	Default
`transparency`	`TransparencyDeclaration \| None`	Optional transparency declaration to embed.	`None`
`risk_assessment`	`RiskAssessment \| None`	Optional risk assessment to embed.	`None`

Returns:

Type	Description
`dict[str, Any]`	Structured documentation dict suitable for submission to
`dict[str, Any]`	national competent authorities.

`RiskAssessment` `dataclass` ¶

AI system risk assessment result (§7.15.1).

EU AI Act Art. 9: Providers must establish a risk management system for the entire lifecycle of the high-risk AI system.

`to_dict()` ¶

Serialise the risk assessment to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation including risk level, category, factors,
`dict[str, Any]`	mitigations, and residual risks.

`RiskClassifier` ¶

Classify AI system risk level per EU AI Act (§7.15.1).

Helps users determine their obligation level based on how they deploy CRP within their AI system.

CRP itself is a context management tool - typically MINIMAL or LIMITED risk. However, if CRP is integrated into a high-risk AI system (e.g., employment screening, credit scoring), the overall system inherits the higher classification.

Usage::

classifier = RiskClassifier()
assessment = classifier.assess(
    category=AISystemCategory.CONTEXT_MANAGEMENT,
    intended_purpose="Managing context for a customer support chatbot",
    processes_personal_data=True,
)
print(f"Risk level: {assessment.risk_level.value}")

`assess(category=AISystemCategory.CONTEXT_MANAGEMENT, intended_purpose='', processes_personal_data=False, makes_automated_decisions=False, affects_fundamental_rights=False, safety_critical=False, profiles_individuals=False)` ¶

Perform risk assessment based on EU AI Act criteria.

Parameters:

Name	Type	Description	Default
`category`	`AISystemCategory`	AI system use-case category.	`CONTEXT_MANAGEMENT`
`intended_purpose`	`str`	Human-readable description of the system's purpose.	`''`
`processes_personal_data`	`bool`	Whether the system processes personal data.	`False`
`makes_automated_decisions`	`bool`	Whether decisions are automated.	`False`
`affects_fundamental_rights`	`bool`	Whether outputs affect fundamental rights.	`False`
`safety_critical`	`bool`	Whether the system is safety-critical.	`False`
`profiles_individuals`	`bool`	Whether individuals are profiled.	`False`

Returns:

Type	Description
`RiskAssessment`	`RiskAssessment` with level, mitigations, and residual risks.

`TransparencyDeclaration` `dataclass` ¶

Transparency declaration for AI system users (§7.15.2).

EU AI Act Art. 13: Providers must ensure that high-risk AI systems are designed and developed in such a way that their operation is sufficiently transparent to enable deployers to interpret the system's output and use it appropriately.

`to_dict()` ¶

Serialise the transparency declaration to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation suitable for disclosure dashboards or regulators.

`ConsentManager` ¶

Manage consent state for data processing purposes (§7.13.1).

EU AI Act Art. 13: Transparency requires clear communication about how data is processed and for what purposes.

Usage::

cm = ConsentManager("session-123")
cm.grant(ProcessingPurpose.ANALYTICS, reason="User opted in")
if cm.check(ProcessingPurpose.ANALYTICS):
    # Process analytics data
    ...
cm.withdraw(ProcessingPurpose.ANALYTICS, reason="User opted out")

`state` `property` ¶

Return the state.

`grant(purpose, reason='', expires_hours=0.0)` ¶

Grant consent for a processing purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being consented to.	required
`reason`	`str`	Human-readable reason for the grant.	`''`
`expires_hours`	`float`	Optional expiry in hours; 0 means no expiry.	`0.0`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`deny(purpose, reason='')` ¶

Deny consent for a processing purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being denied.	required
`reason`	`str`	Human-readable reason for the denial.	`''`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`withdraw(purpose, reason='')` ¶

Withdraw previously granted consent.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being withdrawn.	required
`reason`	`str`	Human-readable reason for the withdrawal.	`''`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`check(purpose)` ¶

Check if consent is granted for a purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted. Required purposes return True by default;
`bool`	opt-in purposes return False until explicitly granted.

`check_required(purpose)` ¶

Check consent and raise if denied for a required purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted.

Raises:

Type	Description
`ConsentRequiredError`	If the purpose is required and consent is missing.

`to_dict()` ¶

Export consent state for compliance reporting.

Returns:

Type	Description
`dict[str, Any]`	Serialised consent state dict.

`HumanOversightController` ¶

Implements human oversight controls (EU AI Act Art. 14) (§7.13.4).

ISO 42001 A.6.2.3: Organizations must establish processes for human oversight of AI systems appropriate to the risk level.

Usage::

hoc = HumanOversightController(OversightConfig(
    level=HumanOversightLevel.APPROVAL,
    require_approval_for_dispatch=True,
))

# Before dispatch:
if hoc.requires_approval("dispatch"):
    approval = hoc.request_approval("dispatch", {"task": "..."})
    # ... wait for human approval ...
    hoc.record_decision(approval.event_id, approved=True)

`config` `property` ¶

Return the config.

`level` `property` ¶

Return the level.

`requires_approval(operation)` ¶

Check if an operation requires human approval.

Parameters:

Name	Type	Description	Default
`operation`	`str`	One of "dispatch", "ingest", "export", "deletion".	required

Returns:

Type	Description
`bool`	True if the configured oversight level and per-operation flags
`bool`	require explicit human approval.

`check_autonomous_limit()` ¶

Check if the autonomous dispatch limit has been reached.

Returns:

Type	Description
`bool`	True if more autonomous dispatches are allowed (or no limit is set).

`record_autonomous_dispatch()` ¶

Record an autonomous dispatch for limit tracking.

Returns:

Type	Description
`None`	None. Increments the internal autonomous dispatch counter.

`request_approval(operation, details=None)` ¶

Create an approval request event.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation requiring approval.	required
`details`	`dict[str, Any] \| None`	Arbitrary context for the reviewer.	`None`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`record_decision(event_id, approved, approved_by='', reason='')` ¶

Record a human oversight decision.

Parameters:

Name	Type	Description	Default
`event_id`	`str`	ID of the original approval request.	required
`approved`	`bool`	True if approved, False if denied.	required
`approved_by`	`str`	Identifier of the reviewer.	`''`
`reason`	`str`	Optional reason for the decision.	`''`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`record_halt(operation, reason, details=None)` ¶

Record a halt event (system stopped due to policy).

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation that was halted.	required
`reason`	`str`	Human-readable reason for the halt.	required
`details`	`dict[str, Any] \| None`	Additional context.	`None`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`should_halt_on_injection()` ¶

Check if processing should halt when injection is detected.

Returns:

Type	Description
`bool`	True if `halt_on_injection_detection` is enabled.

`should_halt_on_pii()` ¶

Check if processing should halt when PII is detected.

Returns:

Type	Description
`bool`	True if `halt_on_pii_detection` is enabled.

`to_dict()` ¶

Export oversight state for compliance reporting.

Returns:

Type	Description
`dict[str, Any]`	Dict summarising oversight level, autonomous dispatch counts, and
`dict[str, Any]`	approval/denial/halt event counts.

`HumanOversightLevel` ¶

Bases: str, Enum

Levels of human oversight (EU AI Act Art. 14) (§7.13.4).

`ProcessingPurpose` ¶

Bases: str, Enum

Data processing purposes - must be declared before processing (§7.13.2).

EU AI Act Art. 10: Data governance requires clear purpose limitation.

`ProcessingRecordKeeper` ¶

Maintain GDPR Article 30 processing records (§7.13.3).

Every data processing activity within a CRP session is recorded with its purpose, legal basis, data categories, and retention.

Usage::

keeper = ProcessingRecordKeeper("session-123")
keeper.record(
    purpose=ProcessingPurpose.FACT_EXTRACTION,
    data_categories=["text_input"],
    legal_basis="legitimate_interest",
    input_size_bytes=4096,
)
records = keeper.export()

`activity_count` `property` ¶

Number of recorded processing activities.

`record(purpose, data_categories, legal_basis='legitimate_interest', input_size_bytes=0, output_size_bytes=0, automated_decision=False, human_oversight=False, retention_period='session')` ¶

Record a data processing activity.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose for the activity.	required
`data_categories`	`list[str]`	Categories of data processed (e.g. ["text_input"]).	required
`legal_basis`	`str`	GDPR legal basis string.	`'legitimate_interest'`
`input_size_bytes`	`int`	Size of input data in bytes.	`0`
`output_size_bytes`	`int`	Size of output data in bytes.	`0`
`automated_decision`	`bool`	Whether automated decision-making occurred.	`False`
`human_oversight`	`bool`	Whether human oversight was available.	`False`
`retention_period`	`str`	Retention descriptor (e.g. "session", "30d").	`'session'`

Returns:

Type	Description
`ProcessingActivity`	The created `ProcessingActivity`.

`export()` ¶

Export all processing records for regulatory review.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts representing every recorded activity.

`summary()` ¶

Summarize processing activities for compliance dashboard.

Returns:

Type	Description
`dict[str, Any]`	Dict with totals, per-purpose counts, input/output byte sums, and
`dict[str, Any]`	oversight/decision counts.

`EmbeddingDefense` ¶

SQ8 quantization + XOR salting for embedding protection (§7.11).

Usage

defense = EmbeddingDefense() protected = defense.protect([0.1, 0.2, -0.3, ...]) recovered = defense.recover(protected)

recovered ≈ original (within quantization error)¶

export_state: embeddings are stripped¶

safe_data = defense.strip_embeddings_for_export(state_dict)

`protect(embedding, salt=None)` ¶

Apply SQ8 quantization + XOR salting (§6H.1, §6H.2).

SQ8: Maps float32 range [min, max] → int8 [-128, 127]. XOR: Applies 4-byte repeating XOR mask to quantized bytes.

`recover(protected)` ¶

Recover embedding from SQ8 + XOR protected form.

Returns approximate original values (quantization introduces error).

`strip_embeddings_for_export(state_dict)` `staticmethod` ¶

Strip all embeddings from state dict for export (§6H.3).

export_state() must export text only - no embeddings.

`ProtectedEmbedding` `dataclass` ¶

Embedding with SQ8 quantization and XOR salt applied.

`to_dict()` ¶

Serialize the protected embedding to a base64-encoded dict.

`from_dict(data)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The data value.	required

Returns:

Type	Description
`ProtectedEmbedding`	`ProtectedEmbedding`.

`EncryptedBlob` `dataclass` ¶

Encrypted data container.

`to_dict()` ¶

Serialize the encrypted blob to a base64-encoded dict.

`from_dict(data)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, str]`	The data value.	required

Returns:

Type	Description
`EncryptedBlob`	`EncryptedBlob`.

`StateEncryptor` ¶

AES-256-GCM encryption for cold state and event logs (§7.3).

Usage

enc = StateEncryptor(session_key) blob = enc.encrypt_cold_state(data_bytes) plaintext = enc.decrypt_cold_state(blob)

`encrypt_cold_state(data)` ¶

Encrypt cold state data (§6C.2).

`decrypt_cold_state(blob)` ¶

Decrypt cold state data (§6C.2).

`encrypt_event_log(data)` ¶

Encrypt event log data (§6C.3).

`decrypt_event_log(blob)` ¶

Decrypt event log data (§6C.3).

`InjectionDetector` ¶

Advisory injection detection - NEVER blocks, only reports (§7.5).

CRITICAL DESIGN CONSTRAINT: This detector NEVER modifies input text and NEVER prevents processing. It only produces advisory flags that are reported to QualityReport.security_flags.

Detection layers (ensembled automatically): Layer 1: Regex pattern library (always active) Layer 2: ML classifier (auto-detected, optional) - prompt-injection-detector: TF-IDF + Logistic Regression (~1MB, MIT) - ProtectAI DeBERTa v2: ONNX transformer (~350MB, Apache 2.0)

Usage

detector = InjectionDetector() report = detector.scan("ignore all previous instructions") if report.has_flags: quality_report.security_flags = report.security_flags

`ml_enabled` `property` ¶

Whether ML-based injection detection is available.

`ml_backend` `property` ¶

Name of the ML backend in use, or 'none'.

`scan(text)` ¶

Scan text for injection patterns.

Returns advisory report - NEVER blocks (§6E.1). Ensembles regex patterns with ML classifier when available.

`InjectionReport` `dataclass` ¶

Result of injection detection - advisory only.

`has_flags` `property` ¶

Return whether this object has flags.

`highest_confidence` `property` ¶

Return the highest confidence.

`security_flags` `property` ¶

Return flag strings for QualityReport.security_flags (§6E.3).

`InjectionType` ¶

Bases: str, Enum

Categories of detected injection patterns.

`FactIntegrityChain` ¶

Tamper-evident chain of fact hashes (§7.7).

Maintains an ordered hash chain. Chain signature is computed via HMAC(session_key, hash_N ‖ ... ‖ hash_0).

Usage

chain = FactIntegrityChain(session_key) chain.add_fact("f1", "capital of France is Paris") sig = chain.chain_signature() assert chain.verify_chain(sig)

`size` `property` ¶

Return the current size count.

`add_fact(fact_id, text)` ¶

Hash a fact and append to the chain.

`get_hash(fact_id)` ¶

Get stored hash for a fact by ID.

`verify_fact(fact_id, text)` ¶

Verify a single fact's hash matches the chain.

`chain_signature()` ¶

Compute HMAC(session_key, hash_N ‖ ... ‖ hash_0) (§6B.3).

Hash chain is concatenated in reverse order (newest first).

`verify_chain(expected_signature)` ¶

Verify the full chain signature matches.

`verify_spot_check(fact_texts, sample_ratio=0.1)` ¶

Spot-check 10% of facts on cold load (§6B.4).

Parameters:

Name	Type	Description	Default
`fact_texts`	`dict[str, str]`	{fact_id: current_text} for verification	required
`sample_ratio`	`float`	fraction of chain to sample (default 10%)	`0.1`

Returns:

Type	Description
`tuple[int, int, list[str]]`	(checked, failures, failed_ids)

`verify_for_envelope(fact_ids, fact_texts)` ¶

Verify all facts before including in envelope (§6B.5).

Returns (all_valid, failed_ids).

`to_dict()` ¶

Serialize the integrity chain entries to a dict.

`export_for_verification()` ¶

Export chain data for external verification (§audit M13).

Returns a dict containing all chain entries and (if session key is set) the chain signature, suitable for independent audit.

`verify_external(export_data, fact_texts)` `staticmethod` ¶

Verify an exported chain against provided fact texts (§audit M13).

This is a static method usable without access to the session key - it re-hashes each fact and compares to the stored hash.

Returns (all_valid, failed_fact_ids).

`from_dict(data, session_key=None)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The data value.	required
`session_key`	`bytes \| None`	The session key value.	`None`

Returns:

Type	Description
`FactIntegrityChain`	`FactIntegrityChain`.

`DataClassification` ¶

Bases: IntEnum

Data sensitivity classification (§7.12.1).

Higher values = more sensitive. Controls encryption, retention, access requirements, and audit verbosity.

`DataLineageTracker` ¶

Track data provenance and transformation history (§7.12.5).

ISO 42001 A.6.2.6: Data management requires tracking data origin, quality, and transformations throughout the AI lifecycle.

Usage::

tracker = DataLineageTracker()
tracker.record("fact-1", "extraction", "user-input", DataClassification.INTERNAL)
tracker.add_transformation("fact-1", "quarantine_promoted")

`record(data_id, origin, source_label='', classification=DataClassification.INTERNAL, parent_ids=None)` ¶

Record a new data lineage entry.

`add_transformation(data_id, transformation)` ¶

Record a transformation applied to data. Returns True if found.

`reclassify(data_id, new_classification)` ¶

Update classification level for a data item. Returns True if found.

`get_lineage(data_id)` ¶

Get lineage entry for a data item.

`get_by_classification(level)` ¶

Get all entries at or above a classification level.

`to_dict()` ¶

Export lineage state for audit/compliance reporting.

`ErasureManager` ¶

Handles right-to-erasure requests (GDPR Article 17) (§7.12.4).

Tracks erasure requests, ensures they are fulfilled, and maintains an audit trail of what was deleted and when.

Usage::

em = ErasureManager()
req = em.create_request("user-hash-abc", scope="session")
# ... caller erases the actual data ...
em.complete_request(req.request_id, items_erased=42)

`create_request(requester_hash, scope='session', target_ids=None)` ¶

Create a new erasure request.

Parameters:

Name	Type	Description	Default
`requester_hash`	`str`	SHA-256 hash of the requester's identity.	required
`scope`	`str`	"session" (all data in current session), "all" (all data across sessions), "specific_facts" (only listed fact IDs).	`'session'`
`target_ids`	`list[str] \| None`	Specific fact IDs when scope="specific_facts".	`None`

Returns:

Type	Description
`ErasureRequest`	ErasureRequest with a unique request_id.

`complete_request(request_id, items_erased=0)` ¶

Mark an erasure request as completed.

Returns True if request was found and completed.

`pending_requests()` ¶

Return all pending (incomplete) erasure requests.

`to_dict()` ¶

Export erasure state for audit/compliance reporting.

`PIIScanner` ¶

Detect PII patterns in text (§7.12.2).

IMPORTANT: This scanner is advisory. It NEVER modifies, redacts, or blocks content (Axiom 9 - output integrity). It reports findings for the user to act upon.

Usage::

scanner = PIIScanner()
result = scanner.scan("Contact john@example.com for details.")
if result.has_pii:
    print(f"Found PII: {result.pii_types_found}")

`scan(text)` ¶

Scan text for PII patterns.

Returns PIIScanResult with detections. Text of matches is NEVER stored - only SHA-256 hashes for audit trail purposes.

`RetentionManager` ¶

Manages data retention and automatic purging (§7.12.3).

EU AI Act Art. 12: Record-keeping with defined retention periods. ISO 42001 A.6.2.8: Records management with lifecycle tracking.

Usage::

rm = RetentionManager()
rm.register("fact-123", DataClassification.RESTRICTED, "user-input")
expired = rm.get_expired()  # Returns IDs ready for purging
rm.mark_purged("fact-123")

`tracked_count` `property` ¶

Return the current tracked count.

`active_count` `property` ¶

Return the current active count.

`register(data_id, classification, source_label='')` ¶

Register a data item for retention tracking.

`get_expired()` ¶

Return IDs of all expired, non-purged data items.

`mark_purged(data_id)` ¶

Mark a data item as purged. Returns True if found.

`enforce()` ¶

Run retention enforcement: return IDs of newly expired items.

The caller is responsible for actually deleting the data. This method only identifies what needs purging.

`get_record(data_id)` ¶

Get retention record for a data item.

`to_dict()` ¶

Export retention state for audit/compliance reporting.

`IngestQuarantine` ¶

1-window quarantine with confidence penalty and batch poisoning detection (§7.8).

Workflow: 1. Incoming facts go into quarantine with 0.7× confidence 2. After 1 window, cross-reference against extraction-derived facts 3. Matching facts are promoted (confidence restored) 4. Non-matching facts are rejected 5. If >30% of a batch fails, quarantine entire batch

Usage

q = IngestQuarantine() q.quarantine_facts([...], "w-1", source="user_input")

... next window processes ...¶

report = q.validate_and_promote("w-2", extraction_fact_texts)

`quarantine_count` `property` ¶

Number of facts currently in quarantine (not promoted or rejected).

`history` `property` ¶

Return the history.

`quarantine_fact(fact_id, original_confidence, window_id, source_label='', fact_text='')` ¶

Place a single fact into quarantine with 0.7× confidence penalty (§6F.1).

`quarantine_facts(facts, window_id, source_label='')` ¶

Quarantine a batch of facts.

Facts can be (fact_id, confidence) or (fact_id, confidence, text).

`get_penalized_confidence(fact_id)` ¶

Get quarantine-penalized confidence for a fact.

`is_quarantined(fact_id)` ¶

Check if a fact is in active quarantine.

`validate_and_promote(current_window_id, extraction_fact_texts, similarity_threshold=0.5)` ¶

Cross-reference validation: promote or reject quarantined facts (§6F.2).

Facts quarantined in an earlier window are validated against extraction-derived facts. Text overlap > threshold → promote.

Parameters:

Name	Type	Description	Default
`current_window_id`	`str`	Current window being processed	required
`extraction_fact_texts`	`dict[str, str]`	{fact_id: text} from extraction pipeline	required
`similarity_threshold`	`float`	Word overlap threshold for cross-reference	`0.5`

Returns:

Type	Description
`QuarantineReport`	QuarantineReport with promotion/rejection counts

`get_active_entries()` ¶

Return all currently quarantined (non-promoted, non-rejected) entries.

`clear()` ¶

Clear all quarantine state.

`QuarantineReport` `dataclass` ¶

Result of cross-reference validation pass.

`Permission` ¶

Bases: str

Named permission strings.

`RateLimitConfig` `dataclass` ¶

Rate limiting configuration (§6G.2).

`RBACEnforcer` ¶

Role-based access control + rate limiting (§7.10).

Usage

rbac = RBACEnforcer(role=Role.OPERATOR) result = rbac.check_permission("dispatch") if not result.allowed: raise PermissionError(result.reason) result = rbac.check_rate_limit("dispatch") if not result.allowed: raise RateLimitError(result.reason) rbac.record_dispatch()

`role` `property` `writable` ¶

Return the role.

`session_tokens_used` `property` ¶

Return the session tokens used.

`check_permission(permission)` ¶

Check if current role has the given permission.

`has_permission(permission)` ¶

Return True if the current role has permission.

`check_rate_limit(operation, payload_bytes=0)` ¶

Check if the operation is within rate limits.

`record_dispatch(tokens_used=0)` ¶

Record a dispatch operation for rate limiting.

`record_ingest(payload_bytes)` ¶

Record an ingest operation for rate limiting.

`record_tokens(count)` ¶

Record token consumption.

`reset_limits()` ¶

Reset all rate limit counters.

`Role` ¶

Bases: IntEnum

RBAC roles with increasing privilege (§6G.1).

`SessionTokenPayload` `dataclass` ¶

Decoded CRP session-token payload (SPEC-007 §2.2).

`to_json_obj()` ¶

Serialize the payload to a JSON-serializable dict.

`from_json_obj(obj)` `classmethod` ¶

Create a new instance from a JSON-compatible object.

Parameters:

Name	Type	Description	Default
`obj`	`dict[str, object]`	The obj value.	required

Returns:

Type	Description
`SessionTokenPayload`	`SessionTokenPayload`.

`TokenStatus` ¶

Bases: str, Enum

Validation outcomes (SPEC-007 §5).

`TokenValidation` `dataclass` ¶

Result of :func:validate_token.

`ok` `property` ¶

Return whether the ok condition holds.

`http_status` `property` ¶

Return the HTTP status.

`CustomSafetyRule` `dataclass` ¶

A user-defined safety rule registered as a first-class citizen (SPEC-033).

Attributes:

Name	Type	Description
`name`	`str`	Unique rule identifier.
`check_fn`	`Any`	Callable that performs the safety check. Type is kept as `Any` to avoid heavy import dependencies in the control plane module.
`description`	`str`	Human-readable explanation of what the rule detects.
`default`	`Any`	Default value for the rule when registered.

`SafetyControlPlane` `dataclass` ¶

Single place from which all CRP safety is seen, tuned, and extended (SPEC-033).

Attributes:

Name	Type	Description
`registry`	`dict[str, SafetyCapability]`	Dict of capability-name → SafetyCapability.
`manifest`	`SafetyManifest`	The SafetyManifest that drives settings.
`coverage`	`SafetyCoverageMap`	The SafetyCoverageMap (capabilities + out-of-scope).
`checkpoints`	`SafetyCoverageMap`	Active checkpoint instances awaiting resolution.

`get_capability(name)` ¶

Retrieve a capability from the registry.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required

Returns:

Type	Description
`SafetyCapability \| None`	The matching `SafetyCapability` or `None` if not registered.

`list_capabilities()` ¶

Return all registered capabilities.

Returns:

Type	Description
`list[SafetyCapability]`	List of every built-in, addable, and custom capability in the registry.

`tune(name, value)` ¶

Change the current value of a capability and sync to Manifest.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required
`value`	`Any`	New current value. Should normally be within `allowed_range`.	required

Returns:

Type	Description
`None`	None. Logs a warning if the capability is unknown.

`register_rule(rule)` ¶

Register a custom safety rule as a first-class citizen.

Parameters:

Name	Type	Description	Default
`rule`	`CustomSafetyRule`	Custom rule definition including name, check function, and default.	required

Returns:

Type	Description
`None`	None. The rule is mirrored into the registry and coverage map.

`create_checkpoint(trigger='always', timeout=300, on_timeout='escalate', on_reject='fallback', context=None)` ¶

Create and track a new human-in-the-loop checkpoint.

Parameters:

Name	Type	Description	Default
`trigger`	`str`	Condition that fires the checkpoint. One of the `CheckpointTrigger` values or a custom string.	`'always'`
`timeout`	`int`	Seconds to wait for human review before auto-resolution.	`300`
`on_timeout`	`str`	Action taken when the checkpoint times out.	`'escalate'`
`on_reject`	`str`	Action taken when the human reviewer rejects.	`'fallback'`
`context`	`dict[str, Any] \| None`	Arbitrary dict passed to the reviewer UI/webhook.	`None`

Returns:

Type	Description
`Checkpoint`	The created `Checkpoint` instance, tracked by ID internally.

`resolve_checkpoint(checkpoint_id, resolution)` ¶

Resolve a checkpoint by ID (called by reviewer or webhook).

Parameters:

Name	Type	Description	Default
`checkpoint_id`	`str`	UUID of the checkpoint returned by `create_checkpoint`.	required
`resolution`	`CheckpointResolution`	Human reviewer's decision and optional edited output.	required

Returns:

Type	Description
`None`	None. Logs an error if the checkpoint ID is unknown.

`get_surface_map()` ¶

Return the complete safety surface as a dict - for UI dashboards.

Returns:

Type	Description
`dict[str, Any]`	Dict with `registry`, `manifest`, `coverage`, and
`dict[str, Any]`	`active_checkpoints` keys.

`show()` ¶

Return a human-readable printout of the entire safety surface.

Returns:

Type	Description
`str`	Multi-line string listing every capability's current/default values,
`str`	allowed ranges, effects, and the explicit out-of-scope list.

`SafetyCapability` `dataclass` ¶

One entry in the Safety Registry / Coverage Map (SPEC-033, SPEC-034).

Attributes:

Name	Type	Description
`name`	`str`	Unique capability identifier.
`description`	`str`	What the capability evaluates.
`spec`	`str`	Specification that defines the capability (e.g. "005", "033").
`default`	`Any`	Factory/default value.
`current`	`Any`	Active value after tuning.
`allowed_range`	`list[Any] \| None`	Optional list or range of permitted values.
`effect`	`str`	Human-readable effect when the capability triggers.
`addable`	`bool`	True if the capability comes from SPEC-034 addable rules.

`to_dict()` ¶

Serialise to dict for dashboard/export.

Returns:

Type	Description
`dict[str, Any]`	JSON-safe dict representation of this capability.

`SafetyCoverageMap` `dataclass` ¶

The complete map of detectable risks + explicit out-of-scope list (SPEC-034).

The out-of-scope list is shown in the Control Plane too - honesty is a feature.

`register(capability)` ¶

Add or overwrite a capability in the map.

Parameters:

Name	Type	Description	Default
`capability`	`SafetyCapability`	Capability descriptor to register.	required

Returns:

Type	Description
`None`	None.

`get(name)` ¶

Lookup a capability by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required

Returns:

Type	Description
`SafetyCapability \| None`	The matching `SafetyCapability` or `None`.

`list_addable()` ¶

Return only the addable rules (SPEC-034).

Returns:

Type	Description
`list[SafetyCapability]`	List of capabilities where `addable` is True.

`to_dict()` ¶

Serialise for dashboard rendering or config export.

Returns:

Type	Description
`dict[str, Any]`	Dict with `capabilities` and `out_of_scope` keys.

`SafetyManifest` `dataclass` ¶

The one config that drives all CRP safety - code and dashboard (SPEC-033 §2).

Attributes:

Name	Type	Description
`profile`	`str`	Named profile ("balanced", "strict", "medical", "financial") or "custom" when any field deviates from a named profile.
`settings`	`dict[str, Any]`	Dict of capability-name → current value.
`checkpoints`	`list[dict[str, Any]]`	List of checkpoint declarations (inline HITL rules).
`custom_rules`	`list[str]`	List of paths to custom rule modules.

`get(name, default=None)` ¶

Read a setting by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Setting key.	required
`default`	`Any`	Value returned if the setting is absent.	`None`

Returns:

Type	Description
`Any`	Current setting value or `default`.

`set(name, value)` ¶

Write a setting - marks profile as custom if it deviates.

Parameters:

Name	Type	Description	Default
`name`	`str`	Setting key.	required
`value`	`Any`	New value to store.	required

Returns:

Type	Description
`None`	None. Updates `profile` to `"custom"` when the value differs
`None`	from the named profile defaults.

`compute_hash()` ¶

Return a deterministic hash of this manifest for CRP-Config-Hash.

Returns:

Type	Description
`str`	Hex-encoded BLAKE2b digest of the canonical settings JSON.

Raises:

Type	Description
`ValueError`	If the settings cannot be serialised (rare).

`to_dict()` ¶

Serialise the manifest to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict with `profile`, `settings`, `checkpoints`, and `custom_rules`.

`from_dict(data)` `classmethod` ¶

Restore a SafetyManifest from a serialised dict.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Dict produced by `to_dict`.	required

Returns:

Type	Description
`SafetyManifest`	Reconstructed `SafetyManifest` instance.

`InputValidator` ¶

Structural input validation - Layer 1, cannot be disabled (§7.4).

This validator ALWAYS runs on all input. It cannot be turned off. It performs structural sanitization without modifying semantic content.

Usage

validator = InputValidator() result = validator.validate("input text") if result.valid: use(result.sanitized_text)

`validate(text, mime_type=None, metadata=None)` ¶

Validate and sanitize input text (§7.4).

Steps: 1. Size check (50 MB limit) 2. Unicode NFC normalization 3. Null byte stripping 4. Control character stripping (keep \n, \t, \r) 5. MIME type validation (if provided) 6. Metadata key count check (≤ 50)

`validate_metadata(metadata)` ¶

Validate and truncate metadata keys (§6D.4).

Returns (sanitized_metadata, warnings).

`ValidationResult` `dataclass` ¶

Result of input validation.

`resolve_clarification(request, handler)` ¶

Resolve a clarification request with a graceful fallback (Invariant 10).

Never raises to the caller: a missing or failing handler degrades to SKIP so the agent continues best-effort rather than crashing the end user's session.

`compute_fact_hash(text)` ¶

Compute integrity hash for a fact's text (§6B.2).

Uses BLAKE3 (~1μs) when available, SHA-256 fallback.

`build_token(payload, master_key)` ¶

Serialise + sign payload, returning payload_b64.signature_b64.

Raises:

Type	Description
`ValueError`	if the encoded payload exceeds 4096 bytes (SPEC-007 §2.4).

`derive_signing_key(master_key, session_id)` ¶

Derive the per-session 256-bit signing key (SPEC-007 §3.1).

`format_set_session_header(token, payload, *, max_age=None)` ¶

Render the CRP-Set-Session header value (SPEC-007 §2.1).

`issue_token(*, session_id, master_key, window=1, quality_history=None, safety_budget=1.0, chain_tip='', continuation_id='', dag_pattern='LINEAR', strategy='', policy_hash='', ckf_hash='', scope='', nonce='', key_version=None, lifetime=DEFAULT_TOKEN_LIFETIME, now=None)` ¶

Convenience: build a fresh payload with iat/exp and sign it.

`parse_token(token)` ¶

Split + decode a token's payload without verifying its signature.

`validate_token(token, *, master_keys, expected_chain_tip=None, expected_scope=None, presented_policy_hash=None, policy_is_more_restrictive=None, presented_nonce=None, now=None)` ¶

Full session-token validation (SPEC-007 §5).

Parameters:

Name	Type	Description	Default
`token`	`str`	the raw `CRP-Session-Token` value.	required
`master_keys`	`bytes \| list[bytes]`	the gateway master key(s). A list allows trying both the current and previous key during master-key rotation (SPEC-007 §6).	required
`expected_chain_tip`	`str \| None`	gateway's recorded HMAC chain tip for the session. Mismatch → 409 (stale/replayed token).	`None`
`expected_scope`	`str \| None`	API-key prefix from the request `Authorization`. Mismatch → 403.	`None`
`presented_policy_hash`	`str \| None`	hash of a `CRP-Safety-Policy` presented with the request. If it differs from the token's `pol` it MUST be more restrictive (set policy_is_more_restrictive) else → 403.	`None`
`policy_is_more_restrictive`	`bool \| None`	caller-computed tightening check.	`None`
`presented_nonce`	`str \| None`	`CRP-Safety-Nonce` from the request. Mismatch → 400.	`None`
`now`	`int \| None`	override current epoch seconds (for testing).	`None`

Returns:

Type	Description
`TokenValidation`	class:`TokenValidation`.

`get_default_control_plane()` ¶

Return the singleton default control plane (lazy initialised).

Returns:

Type	Description
`SafetyControlPlane`	The module-level default `SafetyControlPlane` instance, creating it
`SafetyControlPlane`	on first call.

`security.audit_trail`¶

`crp.security.audit_trail` ¶

Tamper-evident compliance audit trail - HMAC-signed, immutable (§7.14).

Upgrades the observability audit log (§8.9) with: - HMAC-SHA256 chained signatures (tamper-evident) - Compliance-specific event types (data access, consent, erasure) - Structured export for regulatory review - Retention management for audit records - ISO 42001 / EU AI Act Article 12 compliant record-keeping

EU AI Act: Art. 12 (record-keeping), Art. 13 (transparency) ISO 42001: A.6.2.8 (records management), 9.1 (performance monitoring)

`ComplianceEventType` ¶

Bases: str, Enum

Compliance-grade event types beyond standard CRP events (§7.14.1).

`AuditEntry` `dataclass` ¶

Single tamper-evident audit log entry (§7.14.2).

Each entry includes an HMAC-SHA256 signature chained from the previous entry, creating a tamper-evident log. Modification of any entry invalidates all subsequent signatures.

`to_dict()` ¶

Serialise the audit entry to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation including tamper-evidence fields.

`ComplianceAuditTrail` ¶

Tamper-evident, HMAC-signed compliance audit trail (§7.14).

Provides an immutable, append-only log with cryptographic chaining that detects any modification to historical entries.

EU AI Act Art. 12: "High-risk AI systems shall technically allow for the automatic recording of events (logs) over the lifetime of the system."

ISO 42001 A.6.2.8: Organizations must maintain records of AI system development, deployment, operation, monitoring, and decommissioning.

Usage::

trail = ComplianceAuditTrail(signing_key=session_key)
trail.record(
    ComplianceEventType.DATA_INGESTED,
    session_id="abc-123",
    data={"source": "user-input", "size_bytes": 4096},
)
# Verify chain integrity
valid, broken_at = trail.verify_chain()
assert valid
# Export for regulatory review
export = trail.export()

`entry_count` `property` ¶

Number of entries currently stored in the trail.

`record(event_type, session_id='', data=None)` ¶

Append a tamper-evident entry to the audit trail.

Thread-safe. Each entry is chained to the previous via HMAC.

Parameters:

Name	Type	Description	Default
`event_type`	`ComplianceEventType \| str`	Compliance event type enum or custom string.	required
`session_id`	`str`	Session identifier. Falls back to the trail default.	`''`
`data`	`dict[str, Any] \| None`	Arbitrary event payload.	`None`

Returns:

Type	Description
`AuditEntry`	The newly created `AuditEntry`.

`verify_chain()` ¶

Verify the integrity of the entire audit trail.

Returns:

Type	Description
`bool`	`(is_valid, broken_at_sequence)`. `broken_at_sequence` is `-1`
`int`	when the chain is valid, otherwise the sequence of the first broken entry.

`query(*, event_type=None, session_id=None, since=None, until=None)` ¶

Query audit entries with optional filters.

Parameters:

Name	Type	Description	Default
`event_type`	`str \| ComplianceEventType \| None`	Filter by event type.	`None`
`session_id`	`str \| None`	Filter by session ID.	`None`
`since`	`float \| None`	Include entries with timestamp >= this value.	`None`
`until`	`float \| None`	Include entries with timestamp <= this value.	`None`

Returns:

Type	Description
`list[AuditEntry]`	Matching `AuditEntry` objects in chronological order.

`export(*, include_signatures=True, since=None)` ¶

Export the full audit trail for regulatory review.

Parameters:

Name	Type	Description	Default
`include_signatures`	`bool`	Whether to include HMAC signatures in the export.	`True`
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`dict[str, Any]`	Structured document suitable for compliance auditors, including
`dict[str, Any]`	chain integrity status and summary statistics.

`export_jsonl(*, since=None)` ¶

Export as JSONL (one entry per line) for log aggregation.

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`str`	Newline-delimited JSON string of all matching entries.

`export_ndjson(*, since=None)` ¶

Alias for :meth:export_jsonl - NDJSON is the primary format (§4.1).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`str`	Newline-delimited JSON string of all matching entries.

`export_ocsf(*, since=None, provider='', model='')` ¶

Export events in OCSF API Activity format for SIEMs (§4.2).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`
`provider`	`str`	Provider identifier for the destination endpoint field.	`''`
`model`	`str`	Model identifier for the destination endpoint field.	`''`

Returns:

Type	Description
`list[dict[str, Any]]`	List of OCSF `class_uid=6003` records. The HMAC and risk level
`list[dict[str, Any]]`	are carried in the `unmapped` extension.

`export_sarif(*, since=None)` ¶

Export events as a SARIF 2.1.0 log for GitHub integration (§4.3).

Parameters:

Name	Type	Description	Default
`since`	`float \| None`	Only export entries with timestamp >= this value.	`None`

Returns:

Type	Description
`dict[str, Any]`	SARIF 2.1.0 log dict. Non-INFO events become SARIF results; INFO
`dict[str, Any]`	events are emitted at `note` level. Used by CRP Scan (CRP-SPEC-013).

`security.binding`¶

`crp.security.binding` ¶

Session binding - HMAC-SHA256 key derivation and request verification (§7.1).

Provides session-scoped cryptographic binding: - 32-byte session nonce - 256-bit HMAC-derived session key - Constant-time request signature verification - OS keyring storage with zero-config fallback

`SessionBinding` `dataclass` ¶

Cryptographic session binding parameters.

`SessionBindingManager` ¶

HMAC-SHA256 session binding with OS keyring + zero-config fallback (§7.1).

Usage

mgr = SessionBindingManager() binding = mgr.create_session("session-123") sig = mgr.sign_request(b"payload") assert mgr.verify_request_signature(b"payload", sig)

`binding` `property` ¶

Return the binding.

`session_key` `property` ¶

Return current session key (raises if no session).

`key_version` `property` ¶

Current key version number.

`create_session(session_id='')` ¶

Create a new session with fresh nonce and derived key (§6A.1, §6A.2).

session_nonce: 32 bytes of cryptographic randomness
session_key: HMAC-SHA256(master_secret, nonce || session_id)

`sign_request(payload)` ¶

Compute HMAC-SHA256 signature over payload (§6A.3).

`verify_request_signature(payload, signature)` ¶

Constant-time HMAC verification (§6A.3).

Uses hmac.compare_digest to prevent timing attacks. Also checks previous key versions during rotation window.

`rotate_secret()` ¶

Rotate master secret with graceful rollover (§audit H2).

Generates a fresh 256-bit master secret, re-derives the session key, and keeps the previous key(s) valid for verification during the rotation window.

Returns:

Type	Description
`SessionBinding`	Updated SessionBinding with the new key material.

Raises:

Type	Description
`RuntimeError`	If no active session exists.

`store_to_keyring(service_name='crp-sdk')` ¶

Store master secret in OS keyring (DPAPI/Keychain/kernel).

Returns True if stored, False if keyring unavailable.

`from_keyring(service_name='crp-sdk')` `classmethod` ¶

Load master secret from OS keyring. Returns None if unavailable.

`security.checkpoint`¶

`crp.security.checkpoint` ¶

Checkpoint primitive - inline human-in-the-loop for CRP (SPEC-033 §3, SPEC-034).

A Checkpoint is like a breakpoint for human judgment. Declare anywhere:

@client.checkpoint(when="risk >= HIGH")
def generate_medical_advice(prompt):
    ...

Or inline:

result = client.complete(prompt)
await client.checkpoint("approve this output before sending to user")

On timeout or reject, the Checkpoint NEVER leaves the end user with a raw error. It always provides a graceful fallback (Invariant 10).

`CheckpointTrigger` ¶

Bases: str, Enum

What causes a checkpoint to fire (SPEC-033 §3).

`CheckpointTimeoutAction` ¶

Bases: str, Enum

What happens when a checkpoint times out awaiting human response (SPEC-033 §3).

`CheckpointRejectAction` ¶

Bases: str, Enum

What happens when a checkpoint is rejected by the human reviewer (SPEC-033 §3).

`CheckpointResolutionAction` ¶

Bases: str, Enum

The human reviewer's decision (SPEC-033 §3).

`CheckpointResolution` `dataclass` ¶

The result of a human reviewer resolving a checkpoint (SPEC-033 §3).

Attributes:

Name	Type	Description
`action`	`CheckpointResolutionAction`	Reviewer decision (approve/reject/edit).
`reviewer`	`str`	Identifier of the reviewer.
`timestamp`	`float`	Unix timestamp of the resolution.
`edited_output`	`str`	Optional replacement text when action is EDIT.
`audit_event`	`dict[str, Any] \| None`	Optional extra fields for the audit trail.

`to_audit_dict()` ¶

Produce an audit-trail compatible event dict (SPEC-011).

Returns:

Type	Description
`dict[str, Any]`	Dict with checkpoint resolution fields ready for `ComplianceAuditTrail`.

`Checkpoint` `dataclass` ¶

Inline human-in-the-loop declaration (SPEC-033 §3).

Attributes:

Name	Type	Description
`checkpoint_id`	`str`	Unique identifier for this checkpoint instance.
`trigger`	`CheckpointTrigger`	Condition that caused the checkpoint to fire.
`timeout`	`int`	Seconds before auto-action is taken.
`on_timeout`	`CheckpointTimeoutAction`	Action if human does not respond in time.
`on_reject`	`CheckpointRejectAction`	Action if human rejects.
`context`	`dict[str, Any]`	Arbitrary context dict for the reviewer UI.

`wait_for_resolution()` `async` ¶

Block until the checkpoint is resolved by a human reviewer.

If timeout expires, auto-resolves per on_timeout.

Returns:

Type	Description
`CheckpointResolution`	A `CheckpointResolution` - never `None` (Invariant 10).

Raises:

Type	Description
`TimeoutError`	Internally caught and converted to auto-resolution.

`resolve(resolution)` ¶

Called by a human reviewer (or webhook) to resolve this checkpoint.

Parameters:

Name	Type	Description	Default
`resolution`	`CheckpointResolution`	The reviewer's decision and optional edited output.	required

Returns:

Type	Description
`None`	None. Sets the internal resolved event so awaiting tasks unblock.

`decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback')` `classmethod` ¶

Decorator factory: @checkpoint(when="risk >= HIGH").

Parameters:

Name	Type	Description	Default
`when`	`str`	Trigger condition string.	`'always'`
`timeout`	`int`	Seconds to await human review.	`300`
`on_timeout`	`str`	Action if the checkpoint times out.	`'escalate'`
`on_reject`	`str`	Action if the reviewer rejects.	`'fallback'`

Returns:

Type	Description
`Callable`	A decorator that wraps the function with checkpoint gating.

Raises:

Type	Description
`CRPError`	If the reviewer rejects and `on_reject` is `"halt"`.

`security.clarify`¶

`crp.security.clarify` ¶

Clarification bridge - the CLARIFY operation's human-in-the-loop (CRP-SPEC-033/034 + v5).

When a positioned agent does not know something, or is about to use a gated capability, it raises a :class:ClarificationRequest to the user instead of guessing. This is the synchronous companion to the async :class:crp.security.checkpoint.Checkpoint, designed for the positioned-tool-loop.

Invariant 10 (never leave the user with a raw error): resolve_clarification ALWAYS returns a resolution - if no handler is registered, or the handler raises, it degrades gracefully to a SKIP (best-effort continuation), never an exception.

`ClarificationAction` ¶

Bases: str, Enum

The user's response to a clarification request.

`ClarificationRequest` `dataclass` ¶

A request for user input raised by a CLARIFY operation or an oversight gate.

`to_dict()` ¶

Render for streaming / audit / a reviewer UI.

`ClarificationResolution` `dataclass` ¶

The outcome of resolving a clarification request.

`approved` `property` ¶

True when the resolution is an affirmative answer (for oversight gates).

`resolve_clarification(request, handler)` ¶

Resolve a clarification request with a graceful fallback (Invariant 10).

Never raises to the caller: a missing or failing handler degrades to SKIP so the agent continues best-effort rather than crashing the end user's session.

`security.compliance`¶

`crp.security.compliance` ¶

EU AI Act + ISO 42001 compliance framework (§7.15).

Implements

AI system risk classification (EU AI Act Art. 6)
Transparency declarations (EU AI Act Art. 13)
Technical documentation generation (EU AI Act Art. 11)
Compliance status reporting (EU AI Act Art. 9, ISO 42001 9.1)
AI impact assessment (ISO 42001 A.6.2.4)
Quality management system integration (EU AI Act Art. 17)

EU AI Act: Art. 6 (classification), Art. 9-17 (high-risk requirements) ISO 42001: 4-10 (full AIMS lifecycle), A.6.2 (AI-specific controls)

`AIRiskLevel` ¶

Bases: str, Enum

EU AI Act risk classification levels (Art. 6) (§7.15.1).

`AISystemCategory` ¶

Bases: str, Enum

Categories of AI system use cases relevant to risk classification.

`RiskAssessment` `dataclass` ¶

AI system risk assessment result (§7.15.1).

EU AI Act Art. 9: Providers must establish a risk management system for the entire lifecycle of the high-risk AI system.

`to_dict()` ¶

Serialise the risk assessment to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation including risk level, category, factors,
`dict[str, Any]`	mitigations, and residual risks.

`RiskClassifier` ¶

Classify AI system risk level per EU AI Act (§7.15.1).

Helps users determine their obligation level based on how they deploy CRP within their AI system.

CRP itself is a context management tool - typically MINIMAL or LIMITED risk. However, if CRP is integrated into a high-risk AI system (e.g., employment screening, credit scoring), the overall system inherits the higher classification.

Usage::

classifier = RiskClassifier()
assessment = classifier.assess(
    category=AISystemCategory.CONTEXT_MANAGEMENT,
    intended_purpose="Managing context for a customer support chatbot",
    processes_personal_data=True,
)
print(f"Risk level: {assessment.risk_level.value}")

`assess(category=AISystemCategory.CONTEXT_MANAGEMENT, intended_purpose='', processes_personal_data=False, makes_automated_decisions=False, affects_fundamental_rights=False, safety_critical=False, profiles_individuals=False)` ¶

Perform risk assessment based on EU AI Act criteria.

Parameters:

Name	Type	Description	Default
`category`	`AISystemCategory`	AI system use-case category.	`CONTEXT_MANAGEMENT`
`intended_purpose`	`str`	Human-readable description of the system's purpose.	`''`
`processes_personal_data`	`bool`	Whether the system processes personal data.	`False`
`makes_automated_decisions`	`bool`	Whether decisions are automated.	`False`
`affects_fundamental_rights`	`bool`	Whether outputs affect fundamental rights.	`False`
`safety_critical`	`bool`	Whether the system is safety-critical.	`False`
`profiles_individuals`	`bool`	Whether individuals are profiled.	`False`

Returns:

Type	Description
`RiskAssessment`	`RiskAssessment` with level, mitigations, and residual risks.

`TransparencyDeclaration` `dataclass` ¶

Transparency declaration for AI system users (§7.15.2).

EU AI Act Art. 13: Providers must ensure that high-risk AI systems are designed and developed in such a way that their operation is sufficiently transparent to enable deployers to interpret the system's output and use it appropriately.

`to_dict()` ¶

Serialise the transparency declaration to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict representation suitable for disclosure dashboards or regulators.

`ComplianceControl` `dataclass` ¶

Single compliance control status.

`ComplianceReporter` ¶

Generate compliance status reports (§7.15.3).

Maps CRP's native security controls to EU AI Act articles and ISO 42001 clauses, reporting implementation status for each.

Usage::

reporter = ComplianceReporter()
report = reporter.generate_report(session_stats={...})
print(report["summary"]["compliance_score"])

`generate_report(session_stats=None, risk_assessment=None)` ¶

Generate a comprehensive compliance status report.

Parameters:

Name	Type	Description	Default
`session_stats`	`dict[str, Any] \| None`	Optional runtime/session statistics to include.	`None`
`risk_assessment`	`RiskAssessment \| None`	Optional `RiskAssessment` to attach.	`None`

Returns:

Type	Description
`dict[str, Any]`	Dict with EU AI Act and ISO 42001 control lists, implementation
`dict[str, Any]`	counts, compliance percentages, and summary score.

`generate_technical_documentation(transparency=None, risk_assessment=None)` ¶

Generate EU AI Act Art. 11 technical documentation.

Parameters:

Name	Type	Description	Default
`transparency`	`TransparencyDeclaration \| None`	Optional transparency declaration to embed.	`None`
`risk_assessment`	`RiskAssessment \| None`	Optional risk assessment to embed.	`None`

Returns:

Type	Description
`dict[str, Any]`	Structured documentation dict suitable for submission to
`dict[str, Any]`	national competent authorities.

`security.consent`¶

`crp.security.consent` ¶

Consent & data rights management - GDPR + EU AI Act transparency (§7.13).

Implements

Consent state management (opt-in, opt-out, withdrawal)
Purpose limitation (data processing tied to declared purposes)
Processing records (GDPR Article 30 / EU AI Act Art. 12)
Data portability support (export in standard format)
Human oversight controls (EU AI Act Art. 14)

EU AI Act: Art. 13 (transparency), Art. 14 (human oversight), Art. 12 (record-keeping) ISO 42001: A.6.2.3 (human oversight), A.6.2.5 (data collection), A.6.2.7 (data subject rights)

`ConsentStatus` ¶

Bases: str, Enum

Consent states (§7.13.1).

`ProcessingPurpose` ¶

Bases: str, Enum

Data processing purposes - must be declared before processing (§7.13.2).

EU AI Act Art. 10: Data governance requires clear purpose limitation.

`ConsentRecord` `dataclass` ¶

Records a consent decision (§7.13.1).

`ConsentState` `dataclass` ¶

Aggregate consent state for a session.

`is_granted(purpose)` ¶

Check if consent is currently granted for a purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted and not expired. Required purposes
`bool`	return True by default.

`denied_purposes()` ¶

Return purposes that have been explicitly denied or withdrawn.

Returns:

Type	Description
`list[ProcessingPurpose]`	List of purposes with `DENIED` or `WITHDRAWN` status.

`to_dict()` ¶

Serialise the consent state to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict with session id, creation time, and purpose records.

`ConsentManager` ¶

Manage consent state for data processing purposes (§7.13.1).

EU AI Act Art. 13: Transparency requires clear communication about how data is processed and for what purposes.

Usage::

cm = ConsentManager("session-123")
cm.grant(ProcessingPurpose.ANALYTICS, reason="User opted in")
if cm.check(ProcessingPurpose.ANALYTICS):
    # Process analytics data
    ...
cm.withdraw(ProcessingPurpose.ANALYTICS, reason="User opted out")

`state` `property` ¶

Return the state.

`grant(purpose, reason='', expires_hours=0.0)` ¶

Grant consent for a processing purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being consented to.	required
`reason`	`str`	Human-readable reason for the grant.	`''`
`expires_hours`	`float`	Optional expiry in hours; 0 means no expiry.	`0.0`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`deny(purpose, reason='')` ¶

Deny consent for a processing purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being denied.	required
`reason`	`str`	Human-readable reason for the denial.	`''`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`withdraw(purpose, reason='')` ¶

Withdraw previously granted consent.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose being withdrawn.	required
`reason`	`str`	Human-readable reason for the withdrawal.	`''`

Returns:

Type	Description
`ConsentRecord`	The created `ConsentRecord`.

`check(purpose)` ¶

Check if consent is granted for a purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted. Required purposes return True by default;
`bool`	opt-in purposes return False until explicitly granted.

`check_required(purpose)` ¶

Check consent and raise if denied for a required purpose.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose to check.	required

Returns:

Type	Description
`bool`	True if consent is granted.

Raises:

Type	Description
`ConsentRequiredError`	If the purpose is required and consent is missing.

`to_dict()` ¶

Export consent state for compliance reporting.

Returns:

Type	Description
`dict[str, Any]`	Serialised consent state dict.

`ConsentRequiredError` ¶

Bases: Exception

Raised when a required consent is missing.

Required purposes cannot be disabled without degrading core CRP functionality.

`ProcessingActivity` `dataclass` ¶

Records a single data processing activity (GDPR Art. 30) (§7.13.3).

EU AI Act Art. 12: Systems must automatically record events relevant for identifying risks and substantial modifications.

`ProcessingRecordKeeper` ¶

Maintain GDPR Article 30 processing records (§7.13.3).

Every data processing activity within a CRP session is recorded with its purpose, legal basis, data categories, and retention.

Usage::

keeper = ProcessingRecordKeeper("session-123")
keeper.record(
    purpose=ProcessingPurpose.FACT_EXTRACTION,
    data_categories=["text_input"],
    legal_basis="legitimate_interest",
    input_size_bytes=4096,
)
records = keeper.export()

`activity_count` `property` ¶

Number of recorded processing activities.

`record(purpose, data_categories, legal_basis='legitimate_interest', input_size_bytes=0, output_size_bytes=0, automated_decision=False, human_oversight=False, retention_period='session')` ¶

Record a data processing activity.

Parameters:

Name	Type	Description	Default
`purpose`	`ProcessingPurpose`	Processing purpose for the activity.	required
`data_categories`	`list[str]`	Categories of data processed (e.g. ["text_input"]).	required
`legal_basis`	`str`	GDPR legal basis string.	`'legitimate_interest'`
`input_size_bytes`	`int`	Size of input data in bytes.	`0`
`output_size_bytes`	`int`	Size of output data in bytes.	`0`
`automated_decision`	`bool`	Whether automated decision-making occurred.	`False`
`human_oversight`	`bool`	Whether human oversight was available.	`False`
`retention_period`	`str`	Retention descriptor (e.g. "session", "30d").	`'session'`

Returns:

Type	Description
`ProcessingActivity`	The created `ProcessingActivity`.

`export()` ¶

Export all processing records for regulatory review.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts representing every recorded activity.

`summary()` ¶

Summarize processing activities for compliance dashboard.

Returns:

Type	Description
`dict[str, Any]`	Dict with totals, per-purpose counts, input/output byte sums, and
`dict[str, Any]`	oversight/decision counts.

`HumanOversightLevel` ¶

Bases: str, Enum

Levels of human oversight (EU AI Act Art. 14) (§7.13.4).

`OversightConfig` `dataclass` ¶

Human oversight configuration (§7.13.4).

EU AI Act Art. 14: High-risk AI systems must be designed to allow effective human oversight during their period of use.

`OversightEvent` `dataclass` ¶

Records a human oversight event.

`HumanOversightController` ¶

Implements human oversight controls (EU AI Act Art. 14) (§7.13.4).

ISO 42001 A.6.2.3: Organizations must establish processes for human oversight of AI systems appropriate to the risk level.

Usage::

hoc = HumanOversightController(OversightConfig(
    level=HumanOversightLevel.APPROVAL,
    require_approval_for_dispatch=True,
))

# Before dispatch:
if hoc.requires_approval("dispatch"):
    approval = hoc.request_approval("dispatch", {"task": "..."})
    # ... wait for human approval ...
    hoc.record_decision(approval.event_id, approved=True)

`config` `property` ¶

Return the config.

`level` `property` ¶

Return the level.

`requires_approval(operation)` ¶

Check if an operation requires human approval.

Parameters:

Name	Type	Description	Default
`operation`	`str`	One of "dispatch", "ingest", "export", "deletion".	required

Returns:

Type	Description
`bool`	True if the configured oversight level and per-operation flags
`bool`	require explicit human approval.

`check_autonomous_limit()` ¶

Check if the autonomous dispatch limit has been reached.

Returns:

Type	Description
`bool`	True if more autonomous dispatches are allowed (or no limit is set).

`record_autonomous_dispatch()` ¶

Record an autonomous dispatch for limit tracking.

Returns:

Type	Description
`None`	None. Increments the internal autonomous dispatch counter.

`request_approval(operation, details=None)` ¶

Create an approval request event.

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation requiring approval.	required
`details`	`dict[str, Any] \| None`	Arbitrary context for the reviewer.	`None`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`record_decision(event_id, approved, approved_by='', reason='')` ¶

Record a human oversight decision.

Parameters:

Name	Type	Description	Default
`event_id`	`str`	ID of the original approval request.	required
`approved`	`bool`	True if approved, False if denied.	required
`approved_by`	`str`	Identifier of the reviewer.	`''`
`reason`	`str`	Optional reason for the decision.	`''`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`record_halt(operation, reason, details=None)` ¶

Record a halt event (system stopped due to policy).

Parameters:

Name	Type	Description	Default
`operation`	`str`	Operation that was halted.	required
`reason`	`str`	Human-readable reason for the halt.	required
`details`	`dict[str, Any] \| None`	Additional context.	`None`

Returns:

Type	Description
`OversightEvent`	The created `OversightEvent`.

`should_halt_on_injection()` ¶

Check if processing should halt when injection is detected.

Returns:

Type	Description
`bool`	True if `halt_on_injection_detection` is enabled.

`should_halt_on_pii()` ¶

Check if processing should halt when PII is detected.

Returns:

Type	Description
`bool`	True if `halt_on_pii_detection` is enabled.

`to_dict()` ¶

Export oversight state for compliance reporting.

Returns:

Type	Description
`dict[str, Any]`	Dict summarising oversight level, autonomous dispatch counts, and
`dict[str, Any]`	approval/denial/halt event counts.

`security.control_plane`¶

`crp.security.control_plane` ¶

Safety Control Plane (SCP) - single place for all CRP safety (SPEC-033).

The SCP unifies existing scattered safety mechanisms under one catalogue

Registry: every capability, its default, its current setting, its effect
Manifest: the one config that drives code + dashboard
Checkpoints: inline human-in-the-loop declarations
Coverage Map: what CRP detects AND what it explicitly does not detect

Usage::

from crp.security.control_plane import SafetyControlPlane

scp = SafetyControlPlane()
scp.show()                       # human-readable printout
capability = scp.get_capability("require_grounding")
scp.tune("require_grounding", 0.85)
scp.register_rule(my_custom_rule)
surface = scp.get_surface_map()  # for dashboard UI rendering

`CustomSafetyRule` `dataclass` ¶

A user-defined safety rule registered as a first-class citizen (SPEC-033).

Attributes:

Name	Type	Description
`name`	`str`	Unique rule identifier.
`check_fn`	`Any`	Callable that performs the safety check. Type is kept as `Any` to avoid heavy import dependencies in the control plane module.
`description`	`str`	Human-readable explanation of what the rule detects.
`default`	`Any`	Default value for the rule when registered.

`SafetyControlPlane` `dataclass` ¶

Single place from which all CRP safety is seen, tuned, and extended (SPEC-033).

Attributes:

Name	Type	Description
`registry`	`dict[str, SafetyCapability]`	Dict of capability-name → SafetyCapability.
`manifest`	`SafetyManifest`	The SafetyManifest that drives settings.
`coverage`	`SafetyCoverageMap`	The SafetyCoverageMap (capabilities + out-of-scope).
`checkpoints`	`SafetyCoverageMap`	Active checkpoint instances awaiting resolution.

`get_capability(name)` ¶

Retrieve a capability from the registry.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required

Returns:

Type	Description
`SafetyCapability \| None`	The matching `SafetyCapability` or `None` if not registered.

`list_capabilities()` ¶

Return all registered capabilities.

Returns:

Type	Description
`list[SafetyCapability]`	List of every built-in, addable, and custom capability in the registry.

`tune(name, value)` ¶

Change the current value of a capability and sync to Manifest.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required
`value`	`Any`	New current value. Should normally be within `allowed_range`.	required

Returns:

Type	Description
`None`	None. Logs a warning if the capability is unknown.

`register_rule(rule)` ¶

Register a custom safety rule as a first-class citizen.

Parameters:

Name	Type	Description	Default
`rule`	`CustomSafetyRule`	Custom rule definition including name, check function, and default.	required

Returns:

Type	Description
`None`	None. The rule is mirrored into the registry and coverage map.

`create_checkpoint(trigger='always', timeout=300, on_timeout='escalate', on_reject='fallback', context=None)` ¶

Create and track a new human-in-the-loop checkpoint.

Parameters:

Name	Type	Description	Default
`trigger`	`str`	Condition that fires the checkpoint. One of the `CheckpointTrigger` values or a custom string.	`'always'`
`timeout`	`int`	Seconds to wait for human review before auto-resolution.	`300`
`on_timeout`	`str`	Action taken when the checkpoint times out.	`'escalate'`
`on_reject`	`str`	Action taken when the human reviewer rejects.	`'fallback'`
`context`	`dict[str, Any] \| None`	Arbitrary dict passed to the reviewer UI/webhook.	`None`

Returns:

Type	Description
`Checkpoint`	The created `Checkpoint` instance, tracked by ID internally.

`resolve_checkpoint(checkpoint_id, resolution)` ¶

Resolve a checkpoint by ID (called by reviewer or webhook).

Parameters:

Name	Type	Description	Default
`checkpoint_id`	`str`	UUID of the checkpoint returned by `create_checkpoint`.	required
`resolution`	`CheckpointResolution`	Human reviewer's decision and optional edited output.	required

Returns:

Type	Description
`None`	None. Logs an error if the checkpoint ID is unknown.

`get_surface_map()` ¶

Return the complete safety surface as a dict - for UI dashboards.

Returns:

Type	Description
`dict[str, Any]`	Dict with `registry`, `manifest`, `coverage`, and
`dict[str, Any]`	`active_checkpoints` keys.

`show()` ¶

Return a human-readable printout of the entire safety surface.

Returns:

Type	Description
`str`	Multi-line string listing every capability's current/default values,
`str`	allowed ranges, effects, and the explicit out-of-scope list.

`get_default_control_plane()` ¶

Return the singleton default control plane (lazy initialised).

Returns:

Type	Description
`SafetyControlPlane`	The module-level default `SafetyControlPlane` instance, creating it
`SafetyControlPlane`	on first call.

`security.coverage`¶

`crp.security.coverage` ¶

Safety Coverage Map - addable rules registry and explicit out-of-scope list (SPEC-034 §11).

The Coverage Map answers two questions honestly

What risks can CRP detect? (the capabilities)
What risks can CRP NOT detect? (the out-of-scope list)

Honesty is a feature - the Control Plane shows both lists so users have accurate expectations.

`AddableRule` ¶

Metadata for a safety rule that can be registered in the Coverage Map (SPEC-034).

Attributes:

Name	Type	Description
`name`		Unique rule identifier.
`description`		What the rule detects and why it matters.
`default`		Default setting (e.g. "on", "warn").
`allowed_values`		Valid settings for this rule.
`effect`		Human-readable description of the rule's effect when triggered.

`SafetyCapability` `dataclass` ¶

One entry in the Safety Registry / Coverage Map (SPEC-033, SPEC-034).

Attributes:

Name	Type	Description
`name`	`str`	Unique capability identifier.
`description`	`str`	What the capability evaluates.
`spec`	`str`	Specification that defines the capability (e.g. "005", "033").
`default`	`Any`	Factory/default value.
`current`	`Any`	Active value after tuning.
`allowed_range`	`list[Any] \| None`	Optional list or range of permitted values.
`effect`	`str`	Human-readable effect when the capability triggers.
`addable`	`bool`	True if the capability comes from SPEC-034 addable rules.

`to_dict()` ¶

Serialise to dict for dashboard/export.

Returns:

Type	Description
`dict[str, Any]`	JSON-safe dict representation of this capability.

`SafetyCoverageMap` `dataclass` ¶

The complete map of detectable risks + explicit out-of-scope list (SPEC-034).

The out-of-scope list is shown in the Control Plane too - honesty is a feature.

`register(capability)` ¶

Add or overwrite a capability in the map.

Parameters:

Name	Type	Description	Default
`capability`	`SafetyCapability`	Capability descriptor to register.	required

Returns:

Type	Description
`None`	None.

`get(name)` ¶

Lookup a capability by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Capability identifier.	required

Returns:

Type	Description
`SafetyCapability \| None`	The matching `SafetyCapability` or `None`.

`list_addable()` ¶

Return only the addable rules (SPEC-034).

Returns:

Type	Description
`list[SafetyCapability]`	List of capabilities where `addable` is True.

`to_dict()` ¶

Serialise for dashboard rendering or config export.

Returns:

Type	Description
`dict[str, Any]`	Dict with `capabilities` and `out_of_scope` keys.

`security.embedding_defense`¶

`crp.security.embedding_defense` ¶

Embedding defense - SQ8 quantization, XOR salting, no export (§7.11).

Protects stored embeddings from extraction: - SQ8 quantization: float32 → int8 (reduces precision, saves memory) - XOR salting: 4-byte salt per embedding (masks raw values) - No embedding export: export_state() exports text only

`ProtectedEmbedding` `dataclass` ¶

Embedding with SQ8 quantization and XOR salt applied.

`to_dict()` ¶

Serialize the protected embedding to a base64-encoded dict.

`from_dict(data)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The data value.	required

Returns:

Type	Description
`ProtectedEmbedding`	`ProtectedEmbedding`.

`EmbeddingDefense` ¶

SQ8 quantization + XOR salting for embedding protection (§7.11).

Usage

defense = EmbeddingDefense() protected = defense.protect([0.1, 0.2, -0.3, ...]) recovered = defense.recover(protected)

recovered ≈ original (within quantization error)¶

export_state: embeddings are stripped¶

safe_data = defense.strip_embeddings_for_export(state_dict)

`protect(embedding, salt=None)` ¶

Apply SQ8 quantization + XOR salting (§6H.1, §6H.2).

SQ8: Maps float32 range [min, max] → int8 [-128, 127]. XOR: Applies 4-byte repeating XOR mask to quantized bytes.

`recover(protected)` ¶

Recover embedding from SQ8 + XOR protected form.

Returns approximate original values (quantization introduces error).

`strip_embeddings_for_export(state_dict)` `staticmethod` ¶

Strip all embeddings from state dict for export (§6H.3).

export_state() must export text only - no embeddings.

`security.encryption`¶

`crp.security.encryption` ¶

Encryption at rest - AES-256-GCM for cold state and event log (§7.3).

Uses HKDF key diversification so different data classes (cold_storage, event_log) use different derived keys from the same session key.

Requires the cryptography package for production use (AES-256-GCM). A fallback XOR cipher is available for dep-free environments but logs a loud warning - it provides only obfuscation, not real encryption.

`EncryptedBlob` `dataclass` ¶

Encrypted data container.

`to_dict()` ¶

Serialize the encrypted blob to a base64-encoded dict.

`from_dict(data)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, str]`	The data value.	required

Returns:

Type	Description
`EncryptedBlob`	`EncryptedBlob`.

`StateEncryptor` ¶

AES-256-GCM encryption for cold state and event logs (§7.3).

Usage

enc = StateEncryptor(session_key) blob = enc.encrypt_cold_state(data_bytes) plaintext = enc.decrypt_cold_state(blob)

`encrypt_cold_state(data)` ¶

Encrypt cold state data (§6C.2).

`decrypt_cold_state(blob)` ¶

Decrypt cold state data (§6C.2).

`encrypt_event_log(data)` ¶

Encrypt event log data (§6C.3).

`decrypt_event_log(blob)` ¶

Decrypt event log data (§6C.3).

`security.injection`¶

`crp.security.injection` ¶

Injection detection - Layer 2, advisory only, NEVER blocks (§7.5).

Detects prompt injection patterns and reports them as advisory flags. CRITICAL: This detector NEVER modifies input and NEVER blocks processing. It only sets security_flags on the quality report for upstream inspection.

Detection layers (ensembled): 1. Regex pattern library - 21 compiled patterns across 6 categories 2. ML classifier (optional) - prompt-injection-detector (TF-IDF + LR) or ProtectAI DeBERTa v2 (transformer-based). Zero-config: auto-detected.

`InjectionType` ¶

Bases: str, Enum

Categories of detected injection patterns.

`InjectionFlag` `dataclass` ¶

Advisory flag for a detected injection pattern.

`InjectionReport` `dataclass` ¶

Result of injection detection - advisory only.

`has_flags` `property` ¶

Return whether this object has flags.

`highest_confidence` `property` ¶

Return the highest confidence.

`security_flags` `property` ¶

Return flag strings for QualityReport.security_flags (§6E.3).

`InjectionDetector` ¶

Advisory injection detection - NEVER blocks, only reports (§7.5).

CRITICAL DESIGN CONSTRAINT: This detector NEVER modifies input text and NEVER prevents processing. It only produces advisory flags that are reported to QualityReport.security_flags.

Detection layers (ensembled automatically): Layer 1: Regex pattern library (always active) Layer 2: ML classifier (auto-detected, optional) - prompt-injection-detector: TF-IDF + Logistic Regression (~1MB, MIT) - ProtectAI DeBERTa v2: ONNX transformer (~350MB, Apache 2.0)

Usage

detector = InjectionDetector() report = detector.scan("ignore all previous instructions") if report.has_flags: quality_report.security_flags = report.security_flags

`ml_enabled` `property` ¶

Whether ML-based injection detection is available.

`ml_backend` `property` ¶

Name of the ML backend in use, or 'none'.

`scan(text)` ¶

Scan text for injection patterns.

Returns advisory report - NEVER blocks (§6E.1). Ensembles regex patterns with ML classifier when available.

`security.integrity`¶

`crp.security.integrity` ¶

Fact integrity - BLAKE3/SHA-256 hash chain + HMAC verification (§7.2, §7.7).

Provides a tamper-evident chain of fact hashes: - Per-fact hash: BLAKE3 (~1μs) with SHA-256 fallback - Hash chain: ordered sequence of fact hashes - Chain signature: HMAC(session_key, hash_N ‖ ... ‖ hash_0) - Spot-check verification: 10% sample on cold load - Full verification on envelope construction

`ChainEntry` `dataclass` ¶

Single entry in the fact integrity chain.

`FactIntegrityChain` ¶

Tamper-evident chain of fact hashes (§7.7).

Maintains an ordered hash chain. Chain signature is computed via HMAC(session_key, hash_N ‖ ... ‖ hash_0).

Usage

chain = FactIntegrityChain(session_key) chain.add_fact("f1", "capital of France is Paris") sig = chain.chain_signature() assert chain.verify_chain(sig)

`size` `property` ¶

Return the current size count.

`add_fact(fact_id, text)` ¶

Hash a fact and append to the chain.

`get_hash(fact_id)` ¶

Get stored hash for a fact by ID.

`verify_fact(fact_id, text)` ¶

Verify a single fact's hash matches the chain.

`chain_signature()` ¶

Compute HMAC(session_key, hash_N ‖ ... ‖ hash_0) (§6B.3).

Hash chain is concatenated in reverse order (newest first).

`verify_chain(expected_signature)` ¶

Verify the full chain signature matches.

`verify_spot_check(fact_texts, sample_ratio=0.1)` ¶

Spot-check 10% of facts on cold load (§6B.4).

Parameters:

Name	Type	Description	Default
`fact_texts`	`dict[str, str]`	{fact_id: current_text} for verification	required
`sample_ratio`	`float`	fraction of chain to sample (default 10%)	`0.1`

Returns:

Type	Description
`tuple[int, int, list[str]]`	(checked, failures, failed_ids)

`verify_for_envelope(fact_ids, fact_texts)` ¶

Verify all facts before including in envelope (§6B.5).

Returns (all_valid, failed_ids).

`to_dict()` ¶

Serialize the integrity chain entries to a dict.

`export_for_verification()` ¶

Export chain data for external verification (§audit M13).

Returns a dict containing all chain entries and (if session key is set) the chain signature, suitable for independent audit.

`verify_external(export_data, fact_texts)` `staticmethod` ¶

Verify an exported chain against provided fact texts (§audit M13).

This is a static method usable without access to the session key - it re-hashes each fact and compares to the stored hash.

Returns (all_valid, failed_fact_ids).

`from_dict(data, session_key=None)` `classmethod` ¶

Create a new instance from a dictionary.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	The data value.	required
`session_key`	`bytes \| None`	The session key value.	`None`

Returns:

Type	Description
`FactIntegrityChain`	`FactIntegrityChain`.

`compute_fact_hash(text)` ¶

Compute integrity hash for a fact's text (§6B.2).

Uses BLAKE3 (~1μs) when available, SHA-256 fallback.

`security.privacy`¶

`crp.security.privacy` ¶

Privacy controls - data classification, PII detection, retention, erasure (§7.12).

Implements

Data sensitivity classification (PUBLIC → CRITICAL)
PII detection with configurable patterns
Data minimization enforcement
Retention policies with automatic expiry
Right to erasure (GDPR Article 17 / EU AI Act transparency)
Data lineage tracking

EU AI Act: Art. 10 (data governance), Art. 12 (record-keeping), Art. 13 (transparency) ISO 42001: A.6.2.4 (impact assessment), A.6.2.6 (data management), A.6.2.7 (data subject rights)

`DataClassification` ¶

Bases: IntEnum

Data sensitivity classification (§7.12.1).

Higher values = more sensitive. Controls encryption, retention, access requirements, and audit verbosity.

`PIIDetection` `dataclass` ¶

Result of PII scan on a piece of text.

`PIIScanResult` `dataclass` ¶

Aggregate result of PII scanning.

`has_pii` `property` ¶

Return whether this object has PII.

`pii_types_found` `property` ¶

Return the PII types found.

`highest_classification` `property` ¶

Return the highest data classification implied by detected PII.

`to_dict()` ¶

Serialize the PII scan result to a dict.

`PIIScanner` ¶

Detect PII patterns in text (§7.12.2).

IMPORTANT: This scanner is advisory. It NEVER modifies, redacts, or blocks content (Axiom 9 - output integrity). It reports findings for the user to act upon.

Usage::

scanner = PIIScanner()
result = scanner.scan("Contact john@example.com for details.")
if result.has_pii:
    print(f"Found PII: {result.pii_types_found}")

`scan(text)` ¶

Scan text for PII patterns.

Returns PIIScanResult with detections. Text of matches is NEVER stored - only SHA-256 hashes for audit trail purposes.

`RetentionPolicy` `dataclass` ¶

Data retention policy configuration (§7.12.3).

Defines how long different classifications of data are retained and what happens at expiry.

`RetentionRecord` `dataclass` ¶

Tracks retention status for a piece of data.

`RetentionManager` ¶

Manages data retention and automatic purging (§7.12.3).

EU AI Act Art. 12: Record-keeping with defined retention periods. ISO 42001 A.6.2.8: Records management with lifecycle tracking.

Usage::

rm = RetentionManager()
rm.register("fact-123", DataClassification.RESTRICTED, "user-input")
expired = rm.get_expired()  # Returns IDs ready for purging
rm.mark_purged("fact-123")

`tracked_count` `property` ¶

Return the current tracked count.

`active_count` `property` ¶

Return the current active count.

`register(data_id, classification, source_label='')` ¶

Register a data item for retention tracking.

`get_expired()` ¶

Return IDs of all expired, non-purged data items.

`mark_purged(data_id)` ¶

Mark a data item as purged. Returns True if found.

`enforce()` ¶

Run retention enforcement: return IDs of newly expired items.

The caller is responsible for actually deleting the data. This method only identifies what needs purging.

`get_record(data_id)` ¶

Get retention record for a data item.

`to_dict()` ¶

Export retention state for audit/compliance reporting.

`ErasureRequest` `dataclass` ¶

Tracks a right-to-erasure (data deletion) request.

`ErasureManager` ¶

Handles right-to-erasure requests (GDPR Article 17) (§7.12.4).

Tracks erasure requests, ensures they are fulfilled, and maintains an audit trail of what was deleted and when.

Usage::

em = ErasureManager()
req = em.create_request("user-hash-abc", scope="session")
# ... caller erases the actual data ...
em.complete_request(req.request_id, items_erased=42)

`create_request(requester_hash, scope='session', target_ids=None)` ¶

Create a new erasure request.

Parameters:

Name	Type	Description	Default
`requester_hash`	`str`	SHA-256 hash of the requester's identity.	required
`scope`	`str`	"session" (all data in current session), "all" (all data across sessions), "specific_facts" (only listed fact IDs).	`'session'`
`target_ids`	`list[str] \| None`	Specific fact IDs when scope="specific_facts".	`None`

Returns:

Type	Description
`ErasureRequest`	ErasureRequest with a unique request_id.

`complete_request(request_id, items_erased=0)` ¶

Mark an erasure request as completed.

Returns True if request was found and completed.

`pending_requests()` ¶

Return all pending (incomplete) erasure requests.

`to_dict()` ¶

Export erasure state for audit/compliance reporting.

`DataLineageEntry` `dataclass` ¶

Tracks the origin and transformations of a piece of data.

`DataLineageTracker` ¶

Track data provenance and transformation history (§7.12.5).

ISO 42001 A.6.2.6: Data management requires tracking data origin, quality, and transformations throughout the AI lifecycle.

Usage::

tracker = DataLineageTracker()
tracker.record("fact-1", "extraction", "user-input", DataClassification.INTERNAL)
tracker.add_transformation("fact-1", "quarantine_promoted")

`record(data_id, origin, source_label='', classification=DataClassification.INTERNAL, parent_ids=None)` ¶

Record a new data lineage entry.

`add_transformation(data_id, transformation)` ¶

Record a transformation applied to data. Returns True if found.

`reclassify(data_id, new_classification)` ¶

Update classification level for a data item. Returns True if found.

`get_lineage(data_id)` ¶

Get lineage entry for a data item.

`get_by_classification(level)` ¶

Get all entries at or above a classification level.

`to_dict()` ¶

Export lineage state for audit/compliance reporting.

`classification_requirements(level)` ¶

Return minimum security requirements for a data classification level.

`security.quarantine`¶

`crp.security.quarantine` ¶

Ingest quarantine - anti-poisoning with 1-window quarantine (§7.8).

Facts from untrusted sources are quarantined for 1 window with a 0.7× confidence penalty. Cross-reference validation promotes or rejects them. Batch poisoning detection: >30% failures → quarantine entire batch.

`QuarantineEntry` `dataclass` ¶

A fact held in quarantine.

`QuarantineReport` `dataclass` ¶

Result of cross-reference validation pass.

`IngestQuarantine` ¶

1-window quarantine with confidence penalty and batch poisoning detection (§7.8).

Workflow: 1. Incoming facts go into quarantine with 0.7× confidence 2. After 1 window, cross-reference against extraction-derived facts 3. Matching facts are promoted (confidence restored) 4. Non-matching facts are rejected 5. If >30% of a batch fails, quarantine entire batch

Usage

q = IngestQuarantine() q.quarantine_facts([...], "w-1", source="user_input")

... next window processes ...¶

report = q.validate_and_promote("w-2", extraction_fact_texts)

`quarantine_count` `property` ¶

Number of facts currently in quarantine (not promoted or rejected).

`history` `property` ¶

Return the history.

`quarantine_fact(fact_id, original_confidence, window_id, source_label='', fact_text='')` ¶

Place a single fact into quarantine with 0.7× confidence penalty (§6F.1).

`quarantine_facts(facts, window_id, source_label='')` ¶

Quarantine a batch of facts.

Facts can be (fact_id, confidence) or (fact_id, confidence, text).

`get_penalized_confidence(fact_id)` ¶

Get quarantine-penalized confidence for a fact.

`is_quarantined(fact_id)` ¶

Check if a fact is in active quarantine.

`validate_and_promote(current_window_id, extraction_fact_texts, similarity_threshold=0.5)` ¶

Cross-reference validation: promote or reject quarantined facts (§6F.2).

Facts quarantined in an earlier window are validated against extraction-derived facts. Text overlap > threshold → promote.

Parameters:

Name	Type	Description	Default
`current_window_id`	`str`	Current window being processed	required
`extraction_fact_texts`	`dict[str, str]`	{fact_id: text} from extraction pipeline	required
`similarity_threshold`	`float`	Word overlap threshold for cross-reference	`0.5`

Returns:

Type	Description
`QuarantineReport`	QuarantineReport with promotion/rejection counts

`get_active_entries()` ¶

Return all currently quarantined (non-promoted, non-rejected) entries.

`clear()` ¶

Clear all quarantine state.

`security.rbac`¶

`crp.security.rbac` ¶

RBAC & rate limiting - role-based access control + per-session limits (§7.10).

Roles: OBSERVER (read-only), OPERATOR (dispatch + ingest), ADMIN (full). Rate limits: 60 req/min dispatch, 100 MB/min ingest, per-session token cap.

`Role` ¶

Bases: IntEnum

RBAC roles with increasing privilege (§6G.1).

`Permission` ¶

Bases: str

Named permission strings.

`RateLimitConfig` `dataclass` ¶

Rate limiting configuration (§6G.2).

`RateLimitState` `dataclass` ¶

Sliding-window rate limit tracking.

`AccessResult` `dataclass` ¶

Result of an RBAC check.

`RBACEnforcer` ¶

Role-based access control + rate limiting (§7.10).

Usage

rbac = RBACEnforcer(role=Role.OPERATOR) result = rbac.check_permission("dispatch") if not result.allowed: raise PermissionError(result.reason) result = rbac.check_rate_limit("dispatch") if not result.allowed: raise RateLimitError(result.reason) rbac.record_dispatch()

`role` `property` `writable` ¶

Return the role.

`session_tokens_used` `property` ¶

Return the session tokens used.

`check_permission(permission)` ¶

Check if current role has the given permission.

`has_permission(permission)` ¶

Return True if the current role has permission.

`check_rate_limit(operation, payload_bytes=0)` ¶

Check if the operation is within rate limits.

`record_dispatch(tokens_used=0)` ¶

Record a dispatch operation for rate limiting.

`record_ingest(payload_bytes)` ¶

Record an ingest operation for rate limiting.

`record_tokens(count)` ¶

Record token consumption.

`reset_limits()` ¶

Reset all rate limit counters.

`security.safety_manifest`¶

`crp.security.safety_manifest` ¶

Safety Manifest - the single config that drives code + dashboard (SPEC-033 §2).

The Manifest is the one source of truth that both the Safety Control Plane and any dashboard UI read and write. Change it in code and the dashboard reflects it; change it in the dashboard and the file updates.

`SafetyManifest` `dataclass` ¶

The one config that drives all CRP safety - code and dashboard (SPEC-033 §2).

Attributes:

Name	Type	Description
`profile`	`str`	Named profile ("balanced", "strict", "medical", "financial") or "custom" when any field deviates from a named profile.
`settings`	`dict[str, Any]`	Dict of capability-name → current value.
`checkpoints`	`list[dict[str, Any]]`	List of checkpoint declarations (inline HITL rules).
`custom_rules`	`list[str]`	List of paths to custom rule modules.

`get(name, default=None)` ¶

Read a setting by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Setting key.	required
`default`	`Any`	Value returned if the setting is absent.	`None`

Returns:

Type	Description
`Any`	Current setting value or `default`.

`set(name, value)` ¶

Write a setting - marks profile as custom if it deviates.

Parameters:

Name	Type	Description	Default
`name`	`str`	Setting key.	required
`value`	`Any`	New value to store.	required

Returns:

Type	Description
`None`	None. Updates `profile` to `"custom"` when the value differs
`None`	from the named profile defaults.

`compute_hash()` ¶

Return a deterministic hash of this manifest for CRP-Config-Hash.

Returns:

Type	Description
`str`	Hex-encoded BLAKE2b digest of the canonical settings JSON.

Raises:

Type	Description
`ValueError`	If the settings cannot be serialised (rare).

`to_dict()` ¶

Serialise the manifest to a JSON-safe dict.

Returns:

Type	Description
`dict[str, Any]`	Dict with `profile`, `settings`, `checkpoints`, and `custom_rules`.

`from_dict(data)` `classmethod` ¶

Restore a SafetyManifest from a serialised dict.

Parameters:

Name	Type	Description	Default
`data`	`dict[str, Any]`	Dict produced by `to_dict`.	required

Returns:

Type	Description
`SafetyManifest`	Reconstructed `SafetyManifest` instance.

`security.session_token`¶

`crp.security.session_token` ¶

CRP Session Token v3 - signed, stateless session relay (CRP-SPEC-007).

The session token carries signed session state in the CRP-Set-Session response header. The client echoes it back via CRP-Session-Token. Any gateway instance sharing the master key can validate the signature and resume the session - no shared session store required.

Pipeline:

:func:build_token - serialise + HMAC-sign a :class:SessionTokenPayload.
:func:format_set_session_header - render the CRP-Set-Session wire value.
:func:parse_token - split + decode a token without verifying.
:func:validate_token - full validation (signature, expiry, chain tip, scope, policy-hash tightening, nonce) per SPEC-007 §5.

Signing (SPEC-007 §3)::

session_signing_key = HKDF-SHA256(IKM=master_key, salt=sid_bytes,
                                  info="crp-session-sign-v3", L=32)
header    = base64url({"alg":"HS256","typ":"CRP"})
payload   = base64url(json(payload_object))
signature = HMAC-SHA256(header + "." + payload, session_signing_key)
token     = payload + "." + base64url(signature)

`SessionTokenPayload` `dataclass` ¶

Decoded CRP session-token payload (SPEC-007 §2.2).

`to_json_obj()` ¶

Serialize the payload to a JSON-serializable dict.

`from_json_obj(obj)` `classmethod` ¶

Create a new instance from a JSON-compatible object.

Parameters:

Name	Type	Description	Default
`obj`	`dict[str, object]`	The obj value.	required

Returns:

Type	Description
`SessionTokenPayload`	`SessionTokenPayload`.

`TokenStatus` ¶

Bases: str, Enum

Validation outcomes (SPEC-007 §5).

`TokenValidation` `dataclass` ¶

Result of :func:validate_token.

`ok` `property` ¶

Return whether the ok condition holds.

`http_status` `property` ¶

Return the HTTP status.

`derive_signing_key(master_key, session_id)` ¶

Derive the per-session 256-bit signing key (SPEC-007 §3.1).

`build_token(payload, master_key)` ¶

Serialise + sign payload, returning payload_b64.signature_b64.

Raises:

Type	Description
`ValueError`	if the encoded payload exceeds 4096 bytes (SPEC-007 §2.4).

`issue_token(*, session_id, master_key, window=1, quality_history=None, safety_budget=1.0, chain_tip='', continuation_id='', dag_pattern='LINEAR', strategy='', policy_hash='', ckf_hash='', scope='', nonce='', key_version=None, lifetime=DEFAULT_TOKEN_LIFETIME, now=None)` ¶

Convenience: build a fresh payload with iat/exp and sign it.

`format_set_session_header(token, payload, *, max_age=None)` ¶

Render the CRP-Set-Session header value (SPEC-007 §2.1).

`parse_token(token)` ¶

Split + decode a token's payload without verifying its signature.

`validate_token(token, *, master_keys, expected_chain_tip=None, expected_scope=None, presented_policy_hash=None, policy_is_more_restrictive=None, presented_nonce=None, now=None)` ¶

Full session-token validation (SPEC-007 §5).

Parameters:

Name	Type	Description	Default
`token`	`str`	the raw `CRP-Session-Token` value.	required
`master_keys`	`bytes \| list[bytes]`	the gateway master key(s). A list allows trying both the current and previous key during master-key rotation (SPEC-007 §6).	required
`expected_chain_tip`	`str \| None`	gateway's recorded HMAC chain tip for the session. Mismatch → 409 (stale/replayed token).	`None`
`expected_scope`	`str \| None`	API-key prefix from the request `Authorization`. Mismatch → 403.	`None`
`presented_policy_hash`	`str \| None`	hash of a `CRP-Safety-Policy` presented with the request. If it differs from the token's `pol` it MUST be more restrictive (set policy_is_more_restrictive) else → 403.	`None`
`policy_is_more_restrictive`	`bool \| None`	caller-computed tightening check.	`None`
`presented_nonce`	`str \| None`	`CRP-Safety-Nonce` from the request. Mismatch → 400.	`None`
`now`	`int \| None`	override current epoch seconds (for testing).	`None`

Returns:

Type	Description
`TokenValidation`	class:`TokenValidation`.

`security.validation`¶

`crp.security.validation` ¶

Input validation - Layer 1 structural validation, CANNOT be disabled (§7.4).

Enforces: - Size limit: 50 MB - Unicode NFC normalization - Null byte stripping - Control character stripping (preserves \n, \t, \r) - MIME type validation - Metadata key count ≤ 50

`ValidationResult` `dataclass` ¶

Result of input validation.

`InputValidator` ¶

Structural input validation - Layer 1, cannot be disabled (§7.4).

This validator ALWAYS runs on all input. It cannot be turned off. It performs structural sanitization without modifying semantic content.

Usage

validator = InputValidator() result = validator.validate("input text") if result.valid: use(result.sanitized_text)

`validate(text, mime_type=None, metadata=None)` ¶

Validate and sanitize input text (§7.4).

Steps: 1. Size check (50 MB limit) 2. Unicode NFC normalization 3. Null byte stripping 4. Control character stripping (keep \n, \t, \r) 5. MIME type validation (if provided) 6. Metadata key count check (≤ 50)

`validate_metadata(metadata)` ¶

Validate and truncate metadata keys (§6D.4).

Returns (sanitized_metadata, warnings).

crp.security¶

security¶

crp.security ¶

ComplianceAuditTrail ¶

entry_count property ¶

record(event_type, session_id='', data=None) ¶

verify_chain() ¶

query(*, event_type=None, session_id=None, since=None, until=None) ¶

export(*, include_signatures=True, since=None) ¶

export_jsonl(*, since=None) ¶

export_ndjson(*, since=None) ¶

export_ocsf(*, since=None, provider='', model='') ¶

export_sarif(*, since=None) ¶

ComplianceEventType ¶

SessionBindingManager ¶

binding property ¶

session_key property ¶

key_version property ¶

create_session(session_id='') ¶

sign_request(payload) ¶

verify_request_signature(payload, signature) ¶

rotate_secret() ¶

store_to_keyring(service_name='crp-sdk') ¶

from_keyring(service_name='crp-sdk') classmethod ¶

Checkpoint dataclass ¶

wait_for_resolution() async ¶

resolve(resolution) ¶

decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback') classmethod ¶

CheckpointRejectAction ¶

CheckpointResolution dataclass ¶

to_audit_dict() ¶

CheckpointResolutionAction ¶

CheckpointTimeoutAction ¶

CheckpointTrigger ¶

ClarificationAction ¶

ClarificationRequest dataclass ¶

to_dict() ¶

ClarificationResolution dataclass ¶

approved property ¶

AIRiskLevel ¶

AISystemCategory ¶

ComplianceReporter ¶

generate_report(session_stats=None, risk_assessment=None) ¶

generate_technical_documentation(transparency=None, risk_assessment=None) ¶

RiskAssessment dataclass ¶

to_dict() ¶

RiskClassifier ¶

assess(category=AISystemCategory.CONTEXT_MANAGEMENT, intended_purpose='', processes_personal_data=False, makes_automated_decisions=False, affects_fundamental_rights=False, safety_critical=False, profiles_individuals=False) ¶

TransparencyDeclaration dataclass ¶

to_dict() ¶

ConsentManager ¶

state property ¶

grant(purpose, reason='', expires_hours=0.0) ¶

deny(purpose, reason='') ¶

withdraw(purpose, reason='') ¶

check(purpose) ¶

check_required(purpose) ¶

to_dict() ¶

HumanOversightController ¶

config property ¶

level property ¶

requires_approval(operation) ¶

check_autonomous_limit() ¶

record_autonomous_dispatch() ¶

request_approval(operation, details=None) ¶

record_decision(event_id, approved, approved_by='', reason='') ¶

record_halt(operation, reason, details=None) ¶

should_halt_on_injection() ¶

should_halt_on_pii() ¶

to_dict() ¶

HumanOversightLevel ¶

ProcessingPurpose ¶

ProcessingRecordKeeper ¶

activity_count property ¶

record(purpose, data_categories, legal_basis='legitimate_interest', input_size_bytes=0, output_size_bytes=0, automated_decision=False, human_oversight=False, retention_period='session') ¶

export() ¶

summary() ¶

EmbeddingDefense ¶

recovered ≈ original (within quantization error)¶

export_state: embeddings are stripped¶

`security`¶

`crp.security` ¶

`ComplianceAuditTrail` ¶

`entry_count` `property` ¶

`record(event_type, session_id='', data=None)` ¶

`verify_chain()` ¶

`query(*, event_type=None, session_id=None, since=None, until=None)` ¶

`export(*, include_signatures=True, since=None)` ¶

`export_jsonl(*, since=None)` ¶

`export_ndjson(*, since=None)` ¶

`export_ocsf(*, since=None, provider='', model='')` ¶

`export_sarif(*, since=None)` ¶

`ComplianceEventType` ¶

`SessionBindingManager` ¶

`binding` `property` ¶

`session_key` `property` ¶

`key_version` `property` ¶

`create_session(session_id='')` ¶

`sign_request(payload)` ¶

`verify_request_signature(payload, signature)` ¶

`rotate_secret()` ¶

`store_to_keyring(service_name='crp-sdk')` ¶

`from_keyring(service_name='crp-sdk')` `classmethod` ¶

`Checkpoint` `dataclass` ¶

`wait_for_resolution()` `async` ¶

`resolve(resolution)` ¶

`decorate(when='always', timeout=300, on_timeout='escalate', on_reject='fallback')` `classmethod` ¶

`CheckpointRejectAction` ¶

`CheckpointResolution` `dataclass` ¶

`to_audit_dict()` ¶

`CheckpointResolutionAction` ¶

`CheckpointTimeoutAction` ¶

`CheckpointTrigger` ¶

`ClarificationAction` ¶

`ClarificationRequest` `dataclass` ¶

`to_dict()` ¶

`ClarificationResolution` `dataclass` ¶

`approved` `property` ¶

`AIRiskLevel` ¶

`AISystemCategory` ¶

`ComplianceReporter` ¶

`generate_report(session_stats=None, risk_assessment=None)` ¶

`generate_technical_documentation(transparency=None, risk_assessment=None)` ¶

`RiskAssessment` `dataclass` ¶

`to_dict()` ¶

`RiskClassifier` ¶

`assess(category=AISystemCategory.CONTEXT_MANAGEMENT, intended_purpose='', processes_personal_data=False, makes_automated_decisions=False, affects_fundamental_rights=False, safety_critical=False, profiles_individuals=False)` ¶

`TransparencyDeclaration` `dataclass` ¶

`to_dict()` ¶

`ConsentManager` ¶

`state` `property` ¶

`grant(purpose, reason='', expires_hours=0.0)` ¶

`deny(purpose, reason='')` ¶

`withdraw(purpose, reason='')` ¶

`check(purpose)` ¶

`check_required(purpose)` ¶

`to_dict()` ¶

`HumanOversightController` ¶

`config` `property` ¶

`level` `property` ¶

`requires_approval(operation)` ¶

`check_autonomous_limit()` ¶

`record_autonomous_dispatch()` ¶

`request_approval(operation, details=None)` ¶

`record_decision(event_id, approved, approved_by='', reason='')` ¶

`record_halt(operation, reason, details=None)` ¶

`should_halt_on_injection()` ¶

`should_halt_on_pii()` ¶

`to_dict()` ¶

`HumanOversightLevel` ¶

`ProcessingPurpose` ¶

`ProcessingRecordKeeper` ¶

`activity_count` `property` ¶

`record(purpose, data_categories, legal_basis='legitimate_interest', input_size_bytes=0, output_size_bytes=0, automated_decision=False, human_oversight=False, retention_period='session')` ¶

`export()` ¶

`summary()` ¶

`EmbeddingDefense` ¶

`protect(embedding, salt=None)` ¶

`recover(protected)` ¶

`strip_embeddings_for_export(state_dict)` `staticmethod` ¶