Align reference implementation with Draft 04

sovp/core.py

@@ -72,81 +72,133 @@ def sign_identity(private_key_b64: str, identity_metadata: dict) -> str:
     return base64.b64encode(signature).decode('utf-8')
 
 
+def _is_v2_document(identity_metadata: dict) -> bool:
+    context = identity_metadata.get("@context")
+    return isinstance(context, str) and "/v2.0" in context
+
+
+def _parse_timestamp(value: object) -> datetime:
+    if not isinstance(value, str):
+        raise ValueError("timestamp must be a string")
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    if parsed.tzinfo is None:
+        raise ValueError("timestamp must include timezone information")
+    return parsed.astimezone(timezone.utc)
+
+
+def _validate_freshness(
+    identity_metadata: dict,
+    check_timestamp: bool,
+    max_age_seconds: int,
+    clock_skew_seconds: int = 60,
+) -> None:
+    freshness = identity_metadata.get("freshness")
+
+    if _is_v2_document(identity_metadata):
+        if not isinstance(freshness, dict):
+            raise ValueError("v2.0 documents require a freshness object")
+
+        created_str = freshness.get("created")
+        expires_at_str = freshness.get("expiresAt")
+        if created_str is None or expires_at_str is None:
+            raise ValueError(
+                "v2.0 documents require freshness.created and freshness.expiresAt"
+            )
+
+        created = _parse_timestamp(created_str)
+        expires_at = _parse_timestamp(expires_at_str)
+
+        if expires_at <= created:
+            raise ValueError("freshness.expiresAt must be later than freshness.created")
+
+        now = datetime.now(timezone.utc)
+        if now < created - timedelta(seconds=clock_skew_seconds):
+            raise ValueError("freshness.created is too far in the future")
+        if now > expires_at + timedelta(seconds=clock_skew_seconds):
+            raise ValueError("freshness.expiresAt has expired")
+
+        if check_timestamp and now > created + timedelta(
+            seconds=max_age_seconds + clock_skew_seconds
+        ):
+            raise ValueError("freshness.created is outside the issuance window")
+        return
+
+    if check_timestamp:
+        created_str = None
+        if isinstance(freshness, dict):
+            created_str = freshness.get("created")
+        if created_str is None:
+            proof = identity_metadata.get("integrity_proof", {})
+            if isinstance(proof, dict):
+                created_str = proof.get("created")
+        if created_str is None:
+            raise ValueError("created timestamp is required when timestamp checking is enabled")
+
+        created = _parse_timestamp(created_str)
+        now = datetime.now(timezone.utc)
+        if now < created - timedelta(seconds=clock_skew_seconds):
+            raise ValueError("created timestamp is too far in the future")
+        if now > created + timedelta(seconds=max_age_seconds + clock_skew_seconds):
+            raise ValueError("created timestamp is outside the issuance window")
+
+
 def verify_identity(
     identity_metadata: dict,
     signature_b64: str,
     public_key_b64: str,
     check_timestamp: bool = False,
     max_age_seconds: int = 600,
+    expected_host: str | None = None,
 ) -> bool:
     """
-    Deterministically verifies the cryptographic alignment (Psi_core resonance).
-
-    Per draft Section 3: Psi_core = Verify(K_pub, sigma, JCS(M)).
-    Ed25519 pure mode (RFC 8032) is used; no external pre-hash is applied.
-
-    Per draft Section 4 MUST: integrity_proof is stripped before
-    canonicalization, so callers MAY pass the full document or the non-proof
-    subset; the result is identical.
+    Verify an SOVP identity document signature and Draft 04 bindings.
 
-    Per draft Section 7.2 SHOULD: when check_timestamp is True, rejects any
-    document whose created timestamp is older than max_age_seconds (default
-    600). Schema v2.0+: created/expiresAt/nonce live in the signed
-    "freshness" object; this is the primary read location. Schema < v2.0
-    (legacy): falls back to the unsigned integrity_proof.created field for
-    the transition window (draft Section 4 "Migration from v1.4") — this
-    fallback affects only timestamp freshness checking, never signature
-    verification.
+    Schema v2.0 requires freshness.created and freshness.expiresAt. The
+    freshness fields are signed; expiresAt is always enforced, while the
+    created issuance window is enforced when check_timestamp is True.
 
-    Args:
-        identity_metadata (dict): The identity payload (full document or
-            non-proof fields only).
-        signature_b64 (str): The base64 encoded signature.
-        public_key_b64 (str): The base64 encoded Ed25519 public key.
-        check_timestamp (bool): When True, enforce the 600-second validity
-            window on the freshness.created field (draft Section 7.2).
-        max_age_seconds (int): Maximum acceptable age of the created timestamp
-            in seconds. Default: 600 (per draft Section 7.2).
-
-    Returns:
-        bool: True if the signature is valid (Psi_core = 1), False otherwise
-        (Psi_core = 0).
+    When expected_host is supplied, entity.canonical_url MUST resolve to the
+    same host, normalized to lower case with a trailing dot removed.
     """
     try:
+        if expected_host is not None:
+            entity = identity_metadata.get("entity")
+            canonical_url = (
+                entity.get("canonical_url") if isinstance(entity, dict) else None
+            )
+            if not isinstance(canonical_url, str):
+                return False
+            parsed = urlparse(canonical_url)
+            if not parsed.hostname:
+                return False
+            actual_host = parsed.hostname.rstrip(".").lower()
+            wanted_host = expected_host.rstrip(".").lower()
+            if actual_host != wanted_host:
+                return False
+
         pub_bytes = base64.b64decode(public_key_b64)
         public_key = ed25519.Ed25519PublicKey.from_public_bytes(pub_bytes)
         sig_bytes = base64.b64decode(signature_b64)
 
-        # Per draft Section 4 MUST: canonicalize only the non-proof fields.
-        payload = {k: v for k, v in identity_metadata.items() if k not in _OUT_OF_SCOPE_KEYS}
+        payload = {
+            k: v for k, v in identity_metadata.items()
+            if k not in _OUT_OF_SCOPE_KEYS
+        }
         canonical_data = jcs.canonicalize(payload)
         public_key.verify(sig_bytes, canonical_data)
 
-        # Per draft Section 7.2 SHOULD: reject stale timestamps.
-        if check_timestamp:
-            # Schema v2.0+: freshness.created is in the signed scope.
-            # Schema < v2.0 (legacy): fall back to integrity_proof.created,
-            # which is unsigned but is the only location that field exists
-            # in — this fallback only affects timestamp freshness checking,
-            # not signature verification, which already ran above.
-            freshness = identity_metadata.get("freshness", {})
-            created_str = freshness.get("created") if isinstance(freshness, dict) else None
-            if created_str is None:
-                proof = identity_metadata.get("integrity_proof", {})
-                created_str = proof.get("created") if isinstance(proof, dict) else None
-            if created_str is None:
-                return False
-            created = datetime.fromisoformat(created_str.replace("Z", "+00:00"))
-            age = datetime.now(timezone.utc) - created
-            if age > timedelta(seconds=max_age_seconds):
-                return False
-
+        _validate_freshness(
+            identity_metadata,
+            check_timestamp=check_timestamp,
+            max_age_seconds=max_age_seconds,
+        )
         return True
 
     except (InvalidSignature, ValueError, TypeError, AttributeError):
         return False
 
 
+
 def generate_identity_document(
     private_key_b64: str,
     entity_uid: str,
@@ -164,7 +216,7 @@ def generate_identity_document(
     IS part of the signed scope (schema < v2.0 kept them in integrity_proof,
     which is excluded from the signed scope and therefore forgeable without
     invalidating the signature). The scan object (if provided) is appended
-    after integrity_proof and is excluded from the signed scope — consistent
+    after integrity_proof and is excluded from the signed scope. consistent
     with draft V02 item 10: vendor extension objects MUST NOT be included in
     the signed scope.
 
@@ -187,15 +239,19 @@ def generate_identity_document(
     public_key_ref = f"dns:txt:_sovp.{domain}"
     created = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
     resolved_nonce = nonce if nonce is not None else str(uuid.uuid4())
+    resolved_expires_at = expires_at
+    if resolved_expires_at is None:
+        resolved_expires_at = (
+            datetime.now(timezone.utc) + timedelta(hours=1)
+        ).strftime("%Y-%m-%dT%H:%M:%SZ")
 
     freshness = {
         "created": created,
         "nonce": resolved_nonce,
+        "expiresAt": resolved_expires_at,
     }
-    if expires_at is not None:
-        freshness["expiresAt"] = expires_at
 
-    # Non-proof payload — the only fields covered by the signature.
+    # Non-proof payload. the only fields covered by the signature.
     non_proof = {
         "@context": f"https://litzki-systems.com/protocol/{context_version}",
         "@type": "SovereignIdentity",

sovp/resolver.py

@@ -17,36 +17,38 @@ class SOVPResolverError(Exception):
 
 def fetch_identity_document(domain: str, timeout: int = 10) -> dict:
     """
-    Fetch /.well-known/sovp-identity.json from domain.
-    Primary: https://{domain}/.well-known/sovp-identity.json
-    Fallback: https://{domain}/sovp-identity.json
-    Raises SOVPResolverError on HTTP error or JSON parse failure.
-    Returns parsed dict.
+    Fetch the SOVP identity document.
+
+    The well-known endpoint is authoritative. The legacy path is used only
+    when the well-known endpoint returns HTTP 404.
     """
     primary_url = f"https://{domain}/.well-known/sovp-identity.json"
     fallback_url = f"https://{domain}/sovp-identity.json"
     headers = {"User-Agent": "sovp-resolver/1.0.1"}
 
-    for url in (primary_url, fallback_url):
+    try:
+        resp = requests.get(primary_url, timeout=timeout, headers=headers)
+    except requests.RequestException as exc:
+        raise SOVPResolverError(f"failed to retrieve {primary_url}: {exc}") from exc
+
+    url = primary_url
+    if resp.status_code == 404:
+        url = fallback_url
         try:
-            resp = requests.get(url, timeout=timeout, headers=headers)
-            if resp.status_code == 200:
-                # draft-litzki-sovp-04 "Resource Limits for Unverified
-                # Documents": Psi_core is not known yet at this point, so
-                # resp.json() (== json.loads(), no size/depth/duplicate-key
-                # limit) is not used here.
-                try:
-                    return parse_unverified_sovp_document(resp.text)
-                except UnverifiedDocumentLimitError as exc:
-                    raise SOVPResolverError(
-                        f"resource limit violated at {url}: {exc}"
-                    ) from exc
-        except requests.RequestException:
-            continue
+            resp = requests.get(fallback_url, timeout=timeout, headers=headers)
+        except requests.RequestException as exc:
+            raise SOVPResolverError(f"failed to retrieve {fallback_url}: {exc}") from exc
+
+    if resp.status_code != 200:
+        raise SOVPResolverError(f"identity document returned HTTP {resp.status_code} at {url}")
+
+    try:
+        return parse_unverified_sovp_document(resp.text)
+    except UnverifiedDocumentLimitError as exc:
+        raise SOVPResolverError(
+            f"resource limit violated at {url}: {exc}"
+        ) from exc
 
-    raise SOVPResolverError(
-        f"sovp-identity.json not reachable at {primary_url} or {fallback_url}"
-    )
 
 
 def resolve_dns_pubkeys(domain: str, max_keys: int = 4) -> list[str]:
@@ -115,7 +117,7 @@ def validate_domain(domain: str, timeout: int = 10) -> dict:
     """
     Full pipeline:
     1. fetch_identity_document(domain)
-    2. resolve_dns_pubkeys(domain) — every published v=SOVP1 key
+    2. resolve_dns_pubkeys(domain). every published v=SOVP1 key
     3. verify_identity(document, signature, pubkey) against each, in order,
        accepting on the first that verifies (draft-litzki-sovp-04 "Multiple
        _sovp TXT records", key-rotation support)
@@ -134,7 +136,13 @@ def validate_domain(domain: str, timeout: int = 10) -> dict:
 
     psi_core = 0
     for public_key_b64 in candidate_keys:
-        if verify_identity(document, signature_b64, public_key_b64):
+        if verify_identity(
+            document,
+            signature_b64,
+            public_key_b64,
+            check_timestamp=True,
+            expected_host=domain,
+        ):
             psi_core = 1
             break
 

tests/test_core.py

@@ -305,3 +305,71 @@ def test_generate_identity_document_with_scan():
 
     signature = doc["integrity_proof"]["signature"]
     assert verify_identity(doc, signature, pub) is True
+
+
+def _fresh_v2_document(**overrides):
+    from datetime import datetime, timezone, timedelta
+
+    now = datetime.now(timezone.utc)
+    metadata = {
+        "@context": "https://litzki-systems.com/protocol/v2.0",
+        "@type": "SovereignIdentity",
+        "entity": {
+            "uid": "urn:sovp:test-v2",
+            "canonical_url": "https://example.com",
+            "verification_method": "Ed25519",
+        },
+        "freshness": {
+            "created": (now - timedelta(seconds=10)).strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "nonce": "test-nonce",
+            "expiresAt": (now + timedelta(hours=1)).strftime("%Y-%m-%dT%H:%M:%SZ"),
+        },
+    }
+    metadata.update(overrides)
+    return metadata
+
+
+def test_v2_freshness_requires_expires_at():
+    priv, pub = generate_keypair()
+    metadata = _fresh_v2_document()
+    del metadata["freshness"]["expiresAt"]
+    signature = sign_identity(priv, metadata)
+    assert verify_identity(metadata, signature, pub) is False
+
+
+def test_v2_freshness_rejects_expired_document():
+    priv, pub = generate_keypair()
+    metadata = _fresh_v2_document()
+    metadata["freshness"]["expiresAt"] = "2020-01-01T00:00:00Z"
+    signature = sign_identity(priv, metadata)
+    assert verify_identity(metadata, signature, pub) is False
+
+
+def test_v2_freshness_rejects_invalid_expiry_order():
+    priv, pub = generate_keypair()
+    metadata = _fresh_v2_document()
+    metadata["freshness"]["expiresAt"] = metadata["freshness"]["created"]
+    signature = sign_identity(priv, metadata)
+    assert verify_identity(metadata, signature, pub) is False
+
+
+def test_host_binding_accepts_matching_host():
+    priv, pub = generate_keypair()
+    doc = generate_identity_document(
+        private_key_b64=priv,
+        entity_uid="urn:sovp:test-host-match",
+        canonical_url="https://Example.com",
+    )
+    signature = doc["integrity_proof"]["signature"]
+    assert verify_identity(doc, signature, pub, expected_host="example.com") is True
+
+
+def test_host_binding_rejects_mismatched_host():
+    priv, pub = generate_keypair()
+    doc = generate_identity_document(
+        private_key_b64=priv,
+        entity_uid="urn:sovp:test-host-mismatch",
+        canonical_url="https://example.com",
+    )
+    signature = doc["integrity_proof"]["signature"]
+    assert verify_identity(doc, signature, pub, expected_host="attacker.example") is False