From e88a4b3bbb293a99f783efe67f4b7cbc0e3122b8 Mon Sep 17 00:00:00 2001 From: Skull-boy Date: Thu, 10 Sep 2026 12:37:03 +0530 Subject: [PATCH] feat(enforcer): add from_declaration() and cryptographic contract sealing - ContractEnforcer.from_declaration() creates governed enforcer from dict - Validates node_id pattern, T7 wildcard rejection, unknown fields - Cryptographic sealing on both file and declaration contracts - verify_integrity() extended to detect declaration tampering - contract_source and node_id properties added - DeclarationValidationError exported from package root - 25 new tests, 98/98 total passing --- src/scyvera/__init__.py | 2 + src/scyvera/enforcer.py | 262 ++++++++++++++++++++++++- src/scyvera/exceptions.py | 16 ++ tests/test_declaration.py | 389 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 661 insertions(+), 8 deletions(-) create mode 100644 tests/test_declaration.py diff --git a/src/scyvera/__init__.py b/src/scyvera/__init__.py index 5dcf618..699b516 100644 --- a/src/scyvera/__init__.py +++ b/src/scyvera/__init__.py @@ -7,6 +7,7 @@ ContractValidationError, ContractVersionError, ContractViolationError, + DeclarationValidationError, GatewayError, ) from .gateway import BaseGateway, GitHubGateway, QdrantGateway @@ -39,6 +40,7 @@ "ContractVersion", "ContractVersionError", "ContractViolationError", + "DeclarationValidationError", "GatewayError", "GitHubGateway", "LIFECYCLE_DEFAULTS", diff --git a/src/scyvera/enforcer.py b/src/scyvera/enforcer.py index 6d11233..47060d0 100644 --- a/src/scyvera/enforcer.py +++ b/src/scyvera/enforcer.py @@ -54,11 +54,11 @@ import functools import hashlib import inspect +import json import logging from pathlib import Path import re -from typing import Any, Callable, Literal -import warnings +from typing import Any, Callable, Literal, Optional from .exceptions import ( ApprovalPendingError, @@ -66,6 +66,7 @@ ContractValidationError, ContractVersionError, ContractViolationError, + DeclarationValidationError, ) from .validator import ( LIFECYCLE_DEFAULTS, @@ -186,9 +187,11 @@ class ContractEnforcer: def __init__( self, - contract_path: Path, + contract_path: Optional[Path], raw_contract: dict[str, Any], integrity_hash: str, + contract_source: str = "file", + sealed_declaration: Optional[dict[str, Any]] = None, ) -> None: self._contract_path = contract_path self._integrity_hash = integrity_hash @@ -196,6 +199,8 @@ def __init__( self._frozen_contract: Mapping[str, Any] = _deep_freeze(raw_contract) self._audit_log: list[AuditEntry] = [] self._granted_approvals: set[str] = set() + self._contract_source: str = contract_source + self._sealed_declaration: Optional[dict[str, Any]] = sealed_declaration @property def integrity_hash(self) -> str: @@ -208,10 +213,27 @@ def contract(self) -> Mapping[str, Any]: return self._frozen_contract @property - def contract_path(self) -> Path: - """Path to the loaded contract file.""" + def contract_path(self) -> Optional[Path]: + """Path to the loaded contract file. None for declaration contracts.""" return self._contract_path + @property + def contract_source(self) -> str: + """Returns 'file' or 'declaration'.""" + return self._contract_source + + @property + def node_id(self) -> Optional[str]: + """Returns the node_id if declared, None otherwise. + + File-based contracts have no node_id unless they + declare context_contract.node_id — return None for now. + Declaration-based contracts always have a node_id. + """ + if self._contract_source == "declaration": + return self._sealed_declaration.get("node_id") + return None + # ------------------------------------------------------------------------- # Loader and Factory (T5, T7, T8, T10) # ------------------------------------------------------------------------- @@ -266,6 +288,8 @@ def load(cls, path: str | Path) -> ContractEnforcer: contract_path=contract_path, raw_contract=raw_copy, integrity_hash=integrity_hash, + contract_source="file", + sealed_declaration=None, ) @classmethod @@ -342,16 +366,237 @@ def _validate_security_constraints(cls, contract: dict[str, Any], path: Path) -> side_effect_name, ) + # ------------------------------------------------------------------------- + # Declaration-based Factory (Phase 1) + # ------------------------------------------------------------------------- + + _NODE_ID_PATTERN = re.compile(r"^[a-z][a-z0-9-]*-v[0-9]+$") + _VALID_LIFECYCLE_VALUES = frozenset({ + "request-response", "persistent", "scheduled", "ephemeral", "triggered", + }) + _DECLARATION_KNOWN_KEYS = frozenset({ + "node_id", "permissions", "side_effects", "approval_points", + "lifecycle", "state", "recovery_strategy", "observability", + }) + + @classmethod + def from_declaration(cls, declaration: dict[str, Any]) -> ContractEnforcer: + """Create a fully governed ContractEnforcer from a runtime declaration dict. + + The resulting enforcer has identical enforcement properties to a file-based + enforcer. gate() works. audit log works. verify_integrity() works. + The only difference is the source. + + Args: + declaration: A dict describing the contract declaration. Required keys + are ``node_id`` (str matching ``^[a-z][a-z0-9-]*-v[0-9]+$``) and + ``permissions`` (list of str). Optional keys: ``side_effects``, + ``approval_points``, ``lifecycle``, ``state``, ``recovery_strategy``, + ``observability``. + + Returns: + A sealed ContractEnforcer instance with contract_source="declaration". + + Raises: + DeclarationValidationError: If any field is invalid or unknown keys + are present. + """ + # --- Reject unknown keys --- + for key in declaration: + if key not in cls._DECLARATION_KNOWN_KEYS: + raise DeclarationValidationError( + field=key, + reason="unknown field in declaration", + ) + + # --- node_id (required) --- + node_id = declaration.get("node_id") + if node_id is None: + raise DeclarationValidationError( + field="node_id", + reason="missing required field", + ) + if not isinstance(node_id, str): + raise DeclarationValidationError( + field="node_id", + reason=f"must be a string, got {type(node_id).__name__}", + ) + if not cls._NODE_ID_PATTERN.match(node_id): + raise DeclarationValidationError( + field="node_id", + reason=( + f"'{node_id}' does not match required pattern " + f"'^[a-z][a-z0-9-]*-v[0-9]+$' " + f"(examples: 'researcher-v1', 'audit-agent-v2')" + ), + ) + + # --- permissions (required) --- + if "permissions" not in declaration: + raise DeclarationValidationError( + field="permissions", + reason="missing required field", + ) + permissions = declaration["permissions"] + if not isinstance(permissions, list): + raise DeclarationValidationError( + field="permissions", + reason=f"must be a list, got {type(permissions).__name__}", + ) + _WILDCARDS = {"*", "all", "any"} + for i, perm in enumerate(permissions): + if not isinstance(perm, str): + raise DeclarationValidationError( + field="permissions", + reason=f"item at index {i} must be a string, got {type(perm).__name__}", + ) + if not perm: + raise DeclarationValidationError( + field="permissions", + reason=f"item at index {i} is empty — each permission must be a non-empty string", + ) + if perm.strip() in _WILDCARDS or "*" in perm: + raise DeclarationValidationError( + field="permissions", + reason=f"wildcard permission '{perm}' is prohibited (Threat T7)", + ) + + # --- side_effects (optional, default []) --- + side_effects = declaration.get("side_effects", []) + if not isinstance(side_effects, list): + raise DeclarationValidationError( + field="side_effects", + reason=f"must be a list, got {type(side_effects).__name__}", + ) + for i, se in enumerate(side_effects): + if not isinstance(se, str): + raise DeclarationValidationError( + field="side_effects", + reason=f"item at index {i} must be a string, got {type(se).__name__}", + ) + if se.strip() in _WILDCARDS or "*" in se: + raise DeclarationValidationError( + field="side_effects", + reason=f"wildcard side effect '{se}' is prohibited (Threat T7)", + ) + + # --- approval_points (optional, default []) --- + approval_points = declaration.get("approval_points", []) + if not isinstance(approval_points, list): + raise DeclarationValidationError( + field="approval_points", + reason=f"must be a list, got {type(approval_points).__name__}", + ) + for i, ap in enumerate(approval_points): + if not isinstance(ap, (str, dict)): + raise DeclarationValidationError( + field="approval_points", + reason=f"item at index {i} must be a string or dict, got {type(ap).__name__}", + ) + if isinstance(ap, dict) and "before" not in ap and "action" not in ap: + raise DeclarationValidationError( + field="approval_points", + reason=f"dict item at index {i} must have a 'before' or 'action' key", + ) + + # --- lifecycle (optional, default "triggered") --- + lifecycle = declaration.get("lifecycle", "triggered") + if not isinstance(lifecycle, str): + raise DeclarationValidationError( + field="lifecycle", + reason=f"must be a string, got {type(lifecycle).__name__}", + ) + if lifecycle not in cls._VALID_LIFECYCLE_VALUES: + raise DeclarationValidationError( + field="lifecycle", + reason=( + f"'{lifecycle}' is not a valid lifecycle value. " + f"Valid values: {sorted(cls._VALID_LIFECYCLE_VALUES)}" + ), + ) + + # --- state (optional, default "stateless") --- + state = declaration.get("state", "stateless") + if not isinstance(state, str): + raise DeclarationValidationError( + field="state", + reason=f"must be a string, got {type(state).__name__}", + ) + + # --- recovery_strategy (optional) --- + recovery_strategy = declaration.get("recovery_strategy") + if recovery_strategy is not None and not isinstance(recovery_strategy, str): + raise DeclarationValidationError( + field="recovery_strategy", + reason=f"must be a string, got {type(recovery_strategy).__name__}", + ) + + # --- observability (optional) --- + observability = declaration.get("observability") + if observability is not None and not isinstance(observability, dict): + raise DeclarationValidationError( + field="observability", + reason=f"must be a dict, got {type(observability).__name__}", + ) + + # --- Deep copy and seal --- + sealed = copy.deepcopy(declaration) + + content = json.dumps(sealed, sort_keys=True).encode() + integrity_hash = hashlib.sha256(content).hexdigest() + + # Build the internal contract representation to match load() format. + # The enforcer's _is_action_declared() reads from + # self._frozen_contract["permissions"], ["side_effects"], + # ["approval_points"/"approvals"], and ["lifecycle"]. + raw_contract: dict[str, Any] = { + "version": "1.1", + "system": {"name": node_id}, + "permissions": list(permissions), + "side_effects": list(side_effects), + "approval_points": list(approval_points), + "lifecycle": { + "mode": lifecycle if lifecycle in ("request-response", "persistent", "scheduled") else "request-response", + }, + "state": state, + } + if recovery_strategy is not None: + raw_contract["recovery_strategy"] = recovery_strategy + if observability is not None: + raw_contract["observability"] = observability + + return cls( + contract_path=None, + raw_contract=raw_contract, + integrity_hash=integrity_hash, + contract_source="declaration", + sealed_declaration=sealed, + ) + # ------------------------------------------------------------------------- # Integrity Verification (Threat T5) # ------------------------------------------------------------------------- def verify_integrity(self) -> bool: - """Verify that the contract file on disk has not been modified since load time. + """Verify that the contract source has not been modified since load/seal time. + + For file contracts: re-reads the file and compares SHA-256 hash. + For declaration contracts: re-hashes the sealed declaration and compares. Raises: - ContractTamperError: If the file content hash has changed or file is missing. + ContractTamperError: If the content hash has changed or file is missing. """ + if self._contract_source == "declaration": + content = json.dumps(self._sealed_declaration, sort_keys=True).encode() + current_hash = hashlib.sha256(content).hexdigest() + if current_hash != self._integrity_hash: + raise ContractTamperError( + "Declaration contract has been tampered with " + "after sealing" + ) + return True + + # File contract integrity check (existing behavior) if not self._contract_path.exists(): raise ContractTamperError( f"Contract file '{self._contract_path}' was removed from disk after load!" @@ -556,9 +801,10 @@ def _check_approval_required(self, action_name: str) -> tuple[bool, dict[str, An def _log_execution(self, action_name: str, action_type: str, success: bool, error: str | None = None) -> None: """Log execution outcome for observability.""" status = "COMPLETED" if success else f"FAILED: {error}" + source_label = self._contract_path.name if self._contract_path is not None else (self.node_id or "") logger.debug( "Contract execution [%s]: action='%s', type='%s', status='%s'", - self._contract_path.name, + source_label, action_name, action_type, status, diff --git a/src/scyvera/exceptions.py b/src/scyvera/exceptions.py index bcb310f..5563194 100644 --- a/src/scyvera/exceptions.py +++ b/src/scyvera/exceptions.py @@ -96,3 +96,19 @@ def __init__(self, action: str, original_exception: Exception) -> None: super().__init__( f"Gateway error for action '{action}': {original_exception}" ) + + +class DeclarationValidationError(Exception): + """Raised when a runtime contract declaration is invalid. + + This is a creation-time error — raised when from_declaration() + receives a dict that fails validation. Distinct from + ContractViolationError which is a runtime enforcement error. + """ + + def __init__(self, field: str, reason: str) -> None: + self.field = field + self.reason = reason + super().__init__( + f"Invalid contract declaration — field '{field}': {reason}" + ) diff --git a/tests/test_declaration.py b/tests/test_declaration.py new file mode 100644 index 0000000..8d8e42e --- /dev/null +++ b/tests/test_declaration.py @@ -0,0 +1,389 @@ +""" +Comprehensive test suite for Scyvera Declaration-based Contract Enforcement. + +Verifies: +- from_declaration() creation with valid and invalid inputs +- node_id pattern enforcement +- Wildcard rejection (T7) in declarations +- Unknown field rejection +- Deep copy isolation +- Runtime gate enforcement identical to file-based contracts +- Approval gates on declaration-based enforcers +- Audit log correctness for declaration-based enforcers +- Cryptographic sealing and tamper detection for both file and declaration contracts +- contract_source and node_id property correctness +""" +from pathlib import Path + +import pytest + +from scyvera import ( + ApprovalPendingError, + AuditEntry, + ContractEnforcer, + ContractTamperError, + ContractViolationError, + DeclarationValidationError, +) + + +FIXTURES = Path(__file__).resolve().parent / "fixtures" +GITHUB_CONTRACT = FIXTURES / "github_contract.yaml" + + +# ============================================================================= +# 1. Creation Tests +# ============================================================================= + +def test_from_declaration_valid_minimal(): + """Only required fields: node_id, permissions. Must create enforcer without raising.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "researcher-v1", + "permissions": ["qdrant.search"], + }) + assert enforcer is not None + assert enforcer.contract_source == "declaration" + assert enforcer.node_id == "researcher-v1" + assert enforcer.integrity_hash is not None + assert len(enforcer.integrity_hash) == 64 + + +def test_from_declaration_valid_full(): + """All fields present and valid. Must create enforcer without raising.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "audit-agent-v2", + "permissions": ["github.read", "qdrant.search"], + "side_effects": ["github.comment"], + "approval_points": ["github.merge"], + "lifecycle": "ephemeral", + "state": "session", + "recovery_strategy": "retry", + "observability": {"level": "audit", "sinks": ["stdout"]}, + }) + assert enforcer is not None + assert enforcer.contract_source == "declaration" + assert enforcer.node_id == "audit-agent-v2" + + +def test_from_declaration_missing_node_id(): + """DeclarationValidationError, field='node_id'.""" + with pytest.raises(DeclarationValidationError) as exc_info: + ContractEnforcer.from_declaration({ + "permissions": ["github.read"], + }) + assert exc_info.value.field == "node_id" + assert "missing required field" in exc_info.value.reason + + +def test_from_declaration_invalid_node_id_no_version(): + """'researcher' (no -vN suffix) raises DeclarationValidationError.""" + with pytest.raises(DeclarationValidationError) as exc_info: + ContractEnforcer.from_declaration({ + "node_id": "researcher", + "permissions": [], + }) + assert exc_info.value.field == "node_id" + assert "does not match required pattern" in exc_info.value.reason + + +def test_from_declaration_invalid_node_id_uppercase(): + """'Researcher-v1' (uppercase) raises DeclarationValidationError.""" + with pytest.raises(DeclarationValidationError) as exc_info: + ContractEnforcer.from_declaration({ + "node_id": "Researcher-v1", + "permissions": [], + }) + assert exc_info.value.field == "node_id" + assert "does not match required pattern" in exc_info.value.reason + + +def test_from_declaration_invalid_node_id_underscore(): + """'researcher_agent-v1' (underscore) raises DeclarationValidationError.""" + with pytest.raises(DeclarationValidationError) as exc_info: + ContractEnforcer.from_declaration({ + "node_id": "researcher_agent-v1", + "permissions": [], + }) + assert exc_info.value.field == "node_id" + assert "does not match required pattern" in exc_info.value.reason + + +def test_from_declaration_missing_permissions(): + """DeclarationValidationError, field='permissions'.""" + with pytest.raises(DeclarationValidationError) as exc_info: + ContractEnforcer.from_declaration({ + "node_id": "test-v1", + }) + assert exc_info.value.field == "permissions" + assert "missing required field" in exc_info.value.reason + + +def test_from_declaration_wildcard_in_permissions(): + """Wildcards in permissions raise DeclarationValidationError (T7).""" + # Glob-style wildcard + with pytest.raises(DeclarationValidationError) as exc_info: + ContractEnforcer.from_declaration({ + "node_id": "test-v1", + "permissions": ["github.*"], + }) + assert exc_info.value.field == "permissions" + assert "wildcard" in exc_info.value.reason.lower() + + # Bare wildcard + with pytest.raises(DeclarationValidationError) as exc_info: + ContractEnforcer.from_declaration({ + "node_id": "test-v1", + "permissions": ["*"], + }) + assert exc_info.value.field == "permissions" + assert "wildcard" in exc_info.value.reason.lower() + + +def test_from_declaration_unknown_field(): + """Unknown keys raise DeclarationValidationError with the unknown field name.""" + with pytest.raises(DeclarationValidationError) as exc_info: + ContractEnforcer.from_declaration({ + "node_id": "test-v1", + "permissions": [], + "foo": "bar", + }) + assert exc_info.value.field == "foo" + assert "unknown field" in exc_info.value.reason + + +def test_from_declaration_empty_permissions_allowed(): + """permissions=[] is valid — no actions declared, everything is denied by default.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "deny-all-v1", + "permissions": [], + }) + assert enforcer is not None + assert enforcer.node_id == "deny-all-v1" + + +def test_from_declaration_does_not_hold_reference(): + """Deep copy prevents caller mutation from affecting the sealed declaration.""" + declaration = { + "node_id": "safe-agent-v1", + "permissions": ["github.read"], + } + enforcer = ContractEnforcer.from_declaration(declaration) + + # Mutate the original dict after creation + declaration["node_id"] = "TAMPERED" + declaration["permissions"].append("admin.*") + + # verify_integrity must still pass — proves deep copy was made + enforcer.verify_integrity() + assert enforcer.node_id == "safe-agent-v1" + + +# ============================================================================= +# 2. Enforcement Tests (Real ContractEnforcer, Not Mocks) +# ============================================================================= + +def test_declaration_gate_blocks_undeclared_action(): + """Undeclared action raises ContractViolationError.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "test-v1", + "permissions": ["github.read"], + }) + + @enforcer.gate("github.merge", "side_effect") + def merge_pr(): + return "merged" + + with pytest.raises(ContractViolationError) as exc_info: + merge_pr() + assert exc_info.value.action_name == "github.merge" + + +def test_declaration_gate_allows_declared_action(): + """Declared action executes without raising.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "reader-v1", + "permissions": ["github.read"], + "side_effects": ["github.read"], + }) + + @enforcer.gate("github.read", "read") + def read_issue(): + return "issue data" + + result = read_issue() + assert result == "issue data" + + +def test_declaration_gate_blocks_approval_required(): + """Action requiring approval raises ApprovalPendingError.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "deploy-v1", + "permissions": ["prod.deploy"], + "side_effects": ["prod.deploy"], + "approval_points": ["prod.deploy"], + }) + + @enforcer.gate("prod.deploy", "side_effect") + def deploy(): + return "deployed" + + with pytest.raises(ApprovalPendingError) as exc_info: + deploy() + assert exc_info.value.action_name == "prod.deploy" + + +def test_declaration_gate_allows_after_approval_granted(): + """After approval is granted, action executes and audit log shows ALLOWED.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "deploy-v1", + "permissions": ["prod.deploy"], + "side_effects": ["prod.deploy"], + "approval_points": ["prod.deploy"], + }) + + @enforcer.gate("prod.deploy", "side_effect") + def deploy(): + return "deployed" + + # First attempt raises ApprovalPendingError + with pytest.raises(ApprovalPendingError): + deploy() + + # Grant approval + enforcer.approve("prod.deploy", token="AUTH_TOKEN_001") + + # Now it executes + result = deploy() + assert result == "deployed" + + audit = enforcer.get_audit_log() + # PENDING, approval, ALLOWED + allowed_entries = [e for e in audit if e.decision == "ALLOWED"] + assert len(allowed_entries) >= 1 + + +def test_declaration_audit_log_records_denied(): + """Undeclared action produces a DENIED audit entry with correct action name.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "audited-v1", + "permissions": ["github.read"], + }) + + @enforcer.gate("admin.delete", "side_effect") + def delete_everything(): + return "deleted" + + with pytest.raises(ContractViolationError): + delete_everything() + + audit = enforcer.get_audit_log() + assert len(audit) == 1 + assert audit[0].decision == "DENIED" + assert audit[0].action_name == "admin.delete" + + +def test_declaration_audit_log_records_pending(): + """Approval-required action produces a PENDING audit entry.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "pending-v1", + "permissions": ["db.drop"], + "side_effects": ["db.drop"], + "approval_points": ["db.drop"], + }) + + @enforcer.gate("db.drop", "side_effect") + def drop_database(): + return "dropped" + + with pytest.raises(ApprovalPendingError): + drop_database() + + audit = enforcer.get_audit_log() + assert len(audit) == 1 + assert audit[0].decision == "PENDING" + assert audit[0].action_name == "db.drop" + + +# ============================================================================= +# 3. Integrity Tests +# ============================================================================= + +def test_declaration_verify_integrity_passes_on_creation(): + """Fresh declaration passes verify_integrity().""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "fresh-v1", + "permissions": ["github.read"], + }) + assert enforcer.verify_integrity() is True + + +def test_declaration_verify_integrity_detects_tampering(): + """Directly mutating _sealed_declaration triggers ContractTamperError.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "sealed-v1", + "permissions": ["github.read"], + }) + + # Directly mutate the private sealed declaration (simulating memory attack) + enforcer._sealed_declaration["permissions"] = ["admin.*", "root.*"] + + with pytest.raises(ContractTamperError) as exc_info: + enforcer.verify_integrity() + assert "tampered" in str(exc_info.value).lower() + + +def test_file_contract_verify_integrity_passes(): + """File-based contract passes verify_integrity() on unmodified file.""" + enforcer = ContractEnforcer.load(GITHUB_CONTRACT) + assert enforcer.verify_integrity() is True + + +def test_file_contract_verify_integrity_detects_file_change(tmp_path): + """File modification after load triggers ContractTamperError.""" + contract_file = tmp_path / "contract.yaml" + contract_file.write_text( + "version: 1\nworkflow: test\ninputs: []\noutputs: []\npermissions: []\n" + "side_effects: []\napproval_points: []\nrecovery_strategy: retry\n" + "replay_semantics: idempotent\ndependencies: []\nstate: none\nobservability: []\n", + encoding="utf-8", + ) + + enforcer = ContractEnforcer.load(contract_file) + assert enforcer.verify_integrity() is True + + # Modify the file after loading + contract_file.write_text("TAMPERED CONTENT", encoding="utf-8") + + with pytest.raises(ContractTamperError) as exc_info: + enforcer.verify_integrity() + assert "tampered" in str(exc_info.value).lower() + + +def test_contract_source_property_file(): + """File-based contract has contract_source == 'file'.""" + enforcer = ContractEnforcer.load(GITHUB_CONTRACT) + assert enforcer.contract_source == "file" + + +def test_contract_source_property_declaration(): + """Declaration-based contract has contract_source == 'declaration'.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "source-test-v1", + "permissions": [], + }) + assert enforcer.contract_source == "declaration" + + +def test_node_id_property_declaration(): + """Declaration-based contract has the declared node_id.""" + enforcer = ContractEnforcer.from_declaration({ + "node_id": "my-agent-v3", + "permissions": [], + }) + assert enforcer.node_id == "my-agent-v3" + + +def test_node_id_property_file(): + """File-based contract has node_id == None.""" + enforcer = ContractEnforcer.load(GITHUB_CONTRACT) + assert enforcer.node_id is None