From 3078faa8e45dee40d31a351f7946ab9da4efaa46 Mon Sep 17 00:00:00 2001 From: issdandavis <215328633+issdandavis@users.noreply.github.com> Date: Fri, 18 Sep 2026 17:51:40 -0700 Subject: [PATCH 1/3] fix(crypto): authenticate braid vault entries and legacy signatures --- .github/workflows/pqc-native-liboqs.yml | 6 + .../HASH_AND_SIGNATURE_REVIEW_2026-09-18.md | 123 +++++++++ scripts/security/native_liboqs_smoke.py | 23 ++ scripts/system/run_core_python_checks.py | 1 + src/crypto/braid_vault.py | 225 +++++++++------ src/crypto/tri_bundle.py | 9 +- .../scbe_aethermoore/spiral_seal/seal.py | 41 +-- .../spiral_seal/signatures.py | 79 +++--- .../scbe_aethermoore/spiral_seal/seal.py | 261 ++++++++---------- .../spiral_seal/signatures.py | 138 +++++---- .../test_braid_signature_integrity.py | 207 ++++++++++++++ tests/test_braid_vault.py | 6 +- tests/test_industry_grade.py | 15 +- tests/test_scbe_comprehensive.py | 17 +- tests/test_spiral_seal_comprehensive.py | 30 +- 15 files changed, 785 insertions(+), 396 deletions(-) create mode 100644 docs/security/HASH_AND_SIGNATURE_REVIEW_2026-09-18.md create mode 100644 tests/security/test_braid_signature_integrity.py diff --git a/.github/workflows/pqc-native-liboqs.yml b/.github/workflows/pqc-native-liboqs.yml index 00bdd7346..26a460089 100644 --- a/.github/workflows/pqc-native-liboqs.yml +++ b/.github/workflows/pqc-native-liboqs.yml @@ -15,6 +15,9 @@ on: - "tests/crypto/**" - "src/symphonic_cipher/scbe_aethermoore/pqc/**" - "symphonic_cipher/scbe_aethermoore/pqc/**" + - "src/symphonic_cipher/scbe_aethermoore/spiral_seal/**" + - "symphonic_cipher/scbe_aethermoore/spiral_seal/**" + - "tests/security/test_braid_signature_integrity.py" pull_request: branches: [main] paths: @@ -25,6 +28,9 @@ on: - "tests/crypto/**" - "src/symphonic_cipher/scbe_aethermoore/pqc/**" - "symphonic_cipher/scbe_aethermoore/pqc/**" + - "src/symphonic_cipher/scbe_aethermoore/spiral_seal/**" + - "symphonic_cipher/scbe_aethermoore/spiral_seal/**" + - "tests/security/test_braid_signature_integrity.py" concurrency: group: ${{ github.workflow }}-${{ github.head_ref || github.ref }} diff --git a/docs/security/HASH_AND_SIGNATURE_REVIEW_2026-09-18.md b/docs/security/HASH_AND_SIGNATURE_REVIEW_2026-09-18.md new file mode 100644 index 000000000..ff90d8a63 --- /dev/null +++ b/docs/security/HASH_AND_SIGNATURE_REVIEW_2026-09-18.md @@ -0,0 +1,123 @@ +# Hashing, signing and braid composition review + +This is a source review with executable regressions, not a cryptographic proof, +penetration-test certificate, FIPS module validation or NIST certification. + +## Plain-language map + +A hash is a fixed-size fingerprint of input bytes. Matching a trusted fingerprint +helps detect changes. A hash alone cannot identify the sender: anyone can hash +replacement data. HMAC authenticates using a shared secret; a digital signature +uses a private signing key and a public verification key. Encryption hides data; +authenticated encryption also detects tampering. Reversible tokenization is an +encoding, not a cryptographic hash. + +The custom work here mostly composes established primitives or organizes data +around them. That can be useful without claiming a new secure hash primitive. + +| Component | Actual implementation | Role and limit | +| --- | --- | --- | +| `src/crypto/braid_vault.py` | Two domain-separated PBKDF2-HMAC-SHA256 channels, three strands, ordered mixing, final XOR | Experimental key derivation. Neither SHA3+BLAKE2 nor three independent secrets. The word's inverse labels do not invert the rehashed state; no braid-group hardness reduction is established. | +| `src/crypto/tri_bundle.py` | SHA3-256 of ordered packed numeric data, then SHA3-256 of the three bundle hashes and cluster context | Structured identity fingerprints. Each cluster has 27 numeric components and six views have 162. The phi scaling factor is about 207; it is not an entropy or dimension count. | +| `src/crypto/sacred_eggs.py` | SHA-256 shells, HMAC/HKDF-SHA256 and a sorted three-shell binding | Commitments, derivation and contextual binding. Three shells do not imply triple security. Showing a secret to a verifier is not a zero-knowledge proof. | +| `packages/kernel/src/tBraid.ts` | Derived temporal variants, weighted hyperbolic distances and an exponential score | A routing/geometry measure, not a cryptographic hash or signature. | +| `python/scbe/elastic_bijective_hash.py` | Reversible SplitMix64-based mapping with a double-hashed lookup table | Indexing and memory lookup. Not a collision-resistant security primitive. | +| `src/spiralverse/rwp2_envelope.py` | Per-tongue HMAC-SHA256 | Shared-secret authentication, not public-key signatures. Further blockers are listed below. | +| Both `spiral_seal/signatures.py` copies | Native liboqs ML-DSA-65 when available, or legacy Dilithium3; explicitly enabled Ed25519 for development | Actual signing. Ed25519 development mode is classical, not PQ-secure. A standard algorithm alone does not establish FIPS module validation. | + +## Confirmed signing and vault defects repaired + +1. **Format-only authentication:** the legacy fallback accepted any + `FALLBACK_SIG:` prefix followed by 32 bytes, without checking the message or + key. Default key generation/signing now refuses when the native backend is + unavailable. Isolated development mode requires both `SCBE_ALLOW_MOCK_PQC=1` + and `SCBE_ENV=test` or `development`, and uses real Ed25519 verification. + Removing opt-in immediately disables verification of development signatures. +2. **Ignored backend verdict:** the legacy pqcrypto adapter ignored a boolean + false verdict and returned true unless an exception was raised. It now + requires an explicit true result for its legacy CFFI API contract. Unknown + return conventions fail closed and require a separately tested adapter. +3. **Ignored signature request:** the old `seal.py` SS1 wire format omitted KEM + and signature fields; `verify_sig=True` did not verify anything. These + unsupported envelope requests now reject. Unsigned symmetric round trips and + AAD checks remain supported. This class is distinct from the package export + in `spiral_seal.py`; the entire latter protocol is not certified by this review. +4. **Unauthenticated vault entries:** custom XOR encryption could return altered + plaintext. Version 2 uses AES-256-GCM with a fresh 96-bit nonce, a per-entry + derived key and authenticated entry identity, salt, timestamps, tongue and + JSON metadata. Rotation validates all records and stages all new ciphertext + before replacing vault state. TTL zero expires rather than becoming immortal. + +### Compatibility + +Legacy unauthenticated vault entries and format-only signatures are deliberately +rejected. There is no silent downgrade or automatic migration. This in-memory +vault has no documented persistence/export interface; any external retained +legacy ciphertext needs a separate, explicitly reviewed recovery procedure from +a trusted copy. Do not discard old storage on the basis of this change. + +Metadata must be JSON-compatible and finite. Rotation preserves original +creation and expiry timestamps. Replay of a whole valid old entry, concurrent +transactions, secure memory erasure and durable storage are outside this vault's +guarantees. The custom master derivation remains experimental; authenticated +encryption does not prove that derivation secure or repair low-entropy secrets. + +## Remaining blockers and limits + +- RWP2 currently defaults to deterministic, source-visible tongue keys. Its + `|`-joined MAC input is ambiguous for fields containing delimiters, and omits + `kid`, `tier` and `version`. Empty required-tongue sets can pass vacuously. + Do not deploy this legacy path as an authentication boundary until the full + key provisioning and canonical-input contract is repaired and tested. +- Aethercode's `LEDGER SIGN` is a truncated SHA-256 composition with a demo key, + not HMAC or public-key signing. Its execution-proof caller also uses the legacy + RWP2 defaults. Interpreter output must not be treated as security proof. +- `create_session_egg` derives its yolk from the session ID. If that ID is public, + the yolk is public too. It is suitable as a deterministic identifier, not an + independently secret authentication credential. +- Tri-bundle hashes fingerprint packed binary floating-point values. They do + not establish semantic equality across alternate float representations; + field shape, finite-value and canonicalization rules need a separate contract + before using arbitrary external bundles as security commitments. +- Rehashing, additional tongues, nested decimal coordinates and longer encodings + do not create fresh secret entropy. Geometry may route a decision but cannot + authorize a missing or invalid signature. + +## Reproduction and validation + +`tests/security/test_braid_signature_integrity.py` exercises both shipped Python +copies, default/production rejection, explicit development signing, wrong keys, +modified messages, truncated signatures, legacy serializer refusal, vault +header/ciphertext mutation, failed rotation and lifetime preservation. The +initial 16 tests failed on the old implementation. Expanded caller checks also +reproduced ignored verification and ignored backend verdicts before repair. + +`tests/test_braid_vault.py` supplies ordinary vault operation controls. Existing +signing suites now opt in to development mode explicitly and require wrong +message/key rejection instead of merely checking that a boolean was returned. + +Run locally without triggering native library bootstrap: + +```powershell +$env:SCBE_FORCE_SKIP_LIBOQS = '1' +$env:PYTEST_DISABLE_PLUGIN_AUTOLOAD = '1' +python -m pytest -p pytest_asyncio.plugin -o addopts='' tests/security/test_braid_signature_integrity.py tests/test_braid_vault.py -q +``` + +The native liboqs CI smoke separately exercises both real ML-DSA-65 legacy +signing adapters, requiring a valid round trip and rejection of wrong keys, +changed messages, truncation and the old fake prefix. Contract-test doubles are +not evidence of native cryptographic operation. Current results belong in PR +checks and test receipts, rather than being inferred from this document. + +## Primary references + +- [NIST FIPS 202: SHA-3](https://csrc.nist.gov/pubs/fips/202/final). +- [NIST FIPS 204: ML-DSA](https://csrc.nist.gov/pubs/fips/204/final). +- [Cryptography AEAD API and nonce/authentication requirements](https://cryptography.io/en/latest/hazmat/primitives/aead/). +- [Legacy pqcrypto CFFI API](https://github.com/kpdemetriou/pqcrypto), whose + documented verification result is boolean; different generations of this + package require explicit compatibility testing. + +See also [NIST readiness](NIST_READINESS.md) for the broader, partial +SSDF self-assessment and outstanding assurance work. diff --git a/scripts/security/native_liboqs_smoke.py b/scripts/security/native_liboqs_smoke.py index ac1a40712..f9fab009e 100644 --- a/scripts/security/native_liboqs_smoke.py +++ b/scripts/security/native_liboqs_smoke.py @@ -114,6 +114,29 @@ def main() -> int: if wrapper.Dilithium3.verify(keys.public_key, b"tampered fixture", signature): _die(f"{tree}: modified message accepted") + spec = importlib.util.spec_from_file_location( + f"native_spiral_signatures_{index}", root / tree / "scbe_aethermoore/spiral_seal/signatures.py" + ) + signing = importlib.util.module_from_spec(spec) + spec.loader.exec_module(signing) + signing_status = signing.get_pqc_sig_status() + if signing_status["backend"] != "liboqs" or signing_status["algorithm"] != "ML-DSA-65": + _die(f"{tree}: legacy signing API did not select native ML-DSA-65") + sk, pk = signing.dilithium_keygen() + _, wrong_pk = signing.dilithium_keygen() + signed = signing.dilithium_sign(sk, b"native fixture") + if not signing.dilithium_verify(pk, b"native fixture", signed): + _die(f"{tree}: legacy signing API failed its roundtrip") + for test_pk, message, candidate in ( + (pk, b"changed", signed), + (wrong_pk, b"native fixture", signed), + (pk, b"native fixture", signed[:-1]), + (pk, b"native fixture", b"FALLBACK_SIG:" + b"x" * 32), + ): + if signing.dilithium_verify(test_pk, message, candidate): + _die(f"{tree}: legacy signing API accepted invalid authentication") + print(f"legacy_signing_{index}=ML-DSA-65 verified; wrong key/message/truncation/forgery rejected") + print("SCBE_LIBOQS_PASS=1") print("native-liboqs-smoke: PASS") print(f"oqs_module={getattr(oqs, '__file__', 'unknown')}") diff --git a/scripts/system/run_core_python_checks.py b/scripts/system/run_core_python_checks.py index 0069dbe73..c76c81a2d 100644 --- a/scripts/system/run_core_python_checks.py +++ b/scripts/system/run_core_python_checks.py @@ -30,6 +30,7 @@ "tests/governance/test_patent_decision_repairs.py", "tests/governance/test_full_system_decision_precedence.py", "tests/security/test_breathing_and_backend_contracts.py", + "tests/security/test_braid_signature_integrity.py", "tests/test_sacred_eggs.py", "tests/test_sacred_egg_registry.py", "tests/test_semantic_projector_deep.py", diff --git a/src/crypto/braid_vault.py b/src/crypto/braid_vault.py index 7243b868e..7f58b60a6 100644 --- a/src/crypto/braid_vault.py +++ b/src/crypto/braid_vault.py @@ -1,5 +1,5 @@ """ -Braided Dual-Primitive Key Vault (MATHBAC-ready) +Braided key derivation experiment with authenticated vault entries. Architecture: 3 braid strands mapped to tongue pairs: @@ -7,27 +7,31 @@ Strand 1: RU x CA (structure pair) Strand 2: UM x DR (depth pair) - Each strand carries TWO hash primitives (dual): - H_a = SHA3-256 - H_b = BLAKE2b-256 + Each strand carries two domain-separated PBKDF2-HMAC-SHA256 channels, + at 120,000 and 160,000 iterations. They are not independent primitives. - Braid operations (B_3 generators): + Braid-inspired operations (named after B_3 generators): sigma_1: cross strand 0 over strand 1 (mix with XOR + rotate) sigma_2: cross strand 1 over strand 2 - sigma_inv_1, sigma_inv_2: inverse crossings + sigma_inv_1, sigma_inv_2: reverse-labelled mixing operations Key derivation = applying a braid word to the initial 3-strand state. Verification = applying the same braid word and checking equality. - Attack = solving the braid conjugacy problem (computationally hard). + These rehashing operations are not a demonstrated B_3 group action: the + reverse-labelled operations do not invert the state. No conjugacy-hardness, + combined-hash-strength, or post-quantum reduction is claimed. -The braid word itself becomes the "combination lock" — the vault key -is not a string, it's a topological path. +Version 2 entries use AES-256-GCM, binding entry identity and metadata. Legacy +unauthenticated XOR entries are rejected, not silently migrated. Key strength +still depends on secret entropy and the custom derivation needs external review. """ from __future__ import annotations import hashlib import hmac +import json +import math import os import struct import time @@ -35,23 +39,26 @@ from enum import Enum from typing import Any, Dict, List, Optional, Tuple +from cryptography.exceptions import InvalidTag +from cryptography.hazmat.primitives.ciphers.aead import AESGCM + # --------------------------------------------------------------------------- -# Dual hash primitives +# Domain-separated derivation channels # --------------------------------------------------------------------------- def _h_a(data: bytes) -> bytes: - """Primary primitive: PBKDF2-HMAC-SHA256 with a domain-separated salt.""" + """First PBKDF2-HMAC-SHA256 channel with a fixed domain label.""" return hashlib.pbkdf2_hmac("sha256", data, b"braid-vault-h-a", 120_000, dklen=32) def _h_b(data: bytes) -> bytes: - """Secondary primitive: PBKDF2-HMAC-SHA256 with a distinct salt.""" + """Second channel of the SAME primitive with a distinct domain label.""" return hashlib.pbkdf2_hmac("sha256", data, b"braid-vault-h-b", 160_000, dklen=32) def _dual_hash(data: bytes) -> Tuple[bytes, bytes]: - """Hash through both primitives independently.""" + """Derive through both domain-separated channels.""" return _h_a(data), _h_b(data) @@ -82,8 +89,8 @@ def _rotate_left(data: bytes, n: int) -> bytes: class BraidStrand: """One strand carrying dual-hashed state.""" - h_a: bytes # SHA3-256 channel - h_b: bytes # BLAKE2b channel + h_a: bytes # PBKDF2 domain a + h_b: bytes # PBKDF2 domain b def as_bytes(self) -> bytes: return self.h_a + self.h_b @@ -94,7 +101,7 @@ def from_seed(cls, seed: bytes, context: bytes = b"") -> BraidStrand: return cls(h_a=_h_a(keyed), h_b=_h_b(keyed)) def rehash(self) -> BraidStrand: - """Double-hash: feed each channel through the OTHER primitive.""" + """Feed each channel through the other domain's derivation.""" return BraidStrand( h_a=_h_a(self.h_b), # a-channel gets b's output through a h_b=_h_b(self.h_a), # b-channel gets a's output through b @@ -102,7 +109,7 @@ def rehash(self) -> BraidStrand: class BraidCrossing(Enum): - """B_3 generators and their inverses.""" + """Braid-inspired operation labels, not proven invertible generators.""" SIGMA_1 = "s1" # strand 0 over strand 1 SIGMA_2 = "s2" # strand 1 over strand 2 @@ -110,8 +117,8 @@ class BraidCrossing(Enum): SIGMA_2_INV = "s2i" # strand 1 under strand 2 -# The braid relation: sigma_1 * sigma_2 * sigma_1 = sigma_2 * sigma_1 * sigma_2 -# This is NOT commutative: sigma_1 * sigma_2 != sigma_2 * sigma_1 +# Abstract B_3 satisfies the braid relation. This hash-based mixer does not +# establish that relation or inverse state operations; ordering is significant. def _apply_crossing(strands: List[BraidStrand], crossing: BraidCrossing) -> List[BraidStrand]: @@ -120,7 +127,7 @@ def _apply_crossing(strands: List[BraidStrand], crossing: BraidCrossing) -> List Crossing mixes the two involved strands non-commutatively: - Over-strand gets XOR'd with under-strand's dual - Under-strand gets rotated by over-strand's entropy - - Both get rehashed through the opposite primitive + - Both get rehashed through the opposite channel """ s = [BraidStrand(h_a=st.h_a, h_b=st.h_b) for st in strands] @@ -184,9 +191,10 @@ def _apply_crossing(strands: List[BraidStrand], crossing: BraidCrossing) -> List class BraidWord: """A sequence of crossings forming the vault's combination. - The braid word is the KEY. Knowing the word lets you derive the - final state. Without it, you'd need to solve the conjugacy problem - in B_3 to recover the crossing sequence from the final state. + The word and master seed determine the final state. No reduction to + a braid conjugacy problem has been established for this hash mixer. + A generated word of length n has at most 2*n bits of selection entropy; + it does not replace a high-entropy master secret. """ crossings: List[BraidCrossing] = field(default_factory=list) @@ -217,7 +225,7 @@ def generate(cls, length: int = 12) -> BraidWord: return cls(crossings=crossings) def inverse(self) -> BraidWord: - """Compute the inverse braid word (reverse + invert each crossing).""" + """Reverse the symbolic word; this does NOT invert the hashed state.""" inv_map = { BraidCrossing.SIGMA_1: BraidCrossing.SIGMA_1_INV, BraidCrossing.SIGMA_2: BraidCrossing.SIGMA_2_INV, @@ -258,7 +266,7 @@ def _apply_braid(strands: List[BraidStrand], word: BraidWord) -> List[BraidStran def _finalize(strands: List[BraidStrand]) -> bytes: """Collapse 3 strands into a single 32-byte vault key. - Uses triadic mixing: hash all 6 channels (3 strands x 2 primitives) + Uses triadic mixing: hash all 6 channels (3 strands x 2 domains) through a final dual-hash round. """ combined = b"".join(s.as_bytes() for s in strands) # 192 bytes @@ -277,18 +285,20 @@ class VaultEntry: """A single secret stored in the braid vault.""" entry_id: str - ciphertext: bytes # XOR(secret, derived_key) + ciphertext: bytes # AES-GCM ciphertext with authentication tag salt: bytes # random salt mixed into derivation created_at: float expires_at: Optional[float] = None tongue_affinity: str = "KO" # primary tongue for this entry metadata: Dict[str, Any] = field(default_factory=dict) + nonce: bytes = b"" # Empty legacy nonces are explicitly rejected. + format_version: int = 2 @property def is_expired(self) -> bool: if self.expires_at is None: return False - return time.time() > self.expires_at + return time.time() >= self.expires_at # --------------------------------------------------------------------------- @@ -297,23 +307,22 @@ def is_expired(self) -> bool: class BraidVault: - """Braided dual-primitive key vault. + """Custom ordered derivation followed by standard authenticated encryption. The vault derives encryption keys by applying a braid word (the master key) to a 3-strand state initialized from a seed. Each entry gets its own salt, producing a unique derived key per entry. - Security properties: - - Dual primitives: compromise of SHA3 OR BLAKE2b alone is insufficient - - Non-commutative: braid word order matters (topological security) - - Triadic: 3 independent strands = 3 tongue pairs = 192-byte intermediate state - - Conjugacy hardness: recovering the braid word from vault state requires - solving the conjugacy problem in B_3 + Three strands are derived from the SAME master secret, not three independent + secrets. Their 192-byte intermediate state is not 1536 bits of entropy. + Integrity covers the ciphertext and entry metadata; replay of an entire old + valid entry needs external freshness state. This in-memory class does not + provide concurrent transactions, persistence protection or secure erasure. """ def __init__(self, master_seed: bytes, braid_key: BraidWord) -> None: self._master_seed = master_seed - self._braid_key = braid_key + self._braid_key = BraidWord(crossings=list(braid_key.crossings)) self._entries: Dict[str, VaultEntry] = {} self._audit_log: List[Dict[str, Any]] = [] @@ -322,11 +331,69 @@ def __init__(self, master_seed: bytes, braid_key: BraidWord) -> None: self._derived_strands = _apply_braid(init, braid_key) self._vault_key = _finalize(self._derived_strands) - def _derive_entry_key(self, entry_id: str, salt: bytes) -> bytes: + def _derive_entry_key(self, entry_id: str, salt: bytes, vault_key: Optional[bytes] = None) -> bytes: """Derive a per-entry encryption key from the vault key.""" - material = self._vault_key + entry_id.encode("utf-8") + salt + encoded_id = entry_id.encode("utf-8") + material = b"scbe:braid-entry:v2\0" + (self._vault_key if vault_key is None else vault_key) + material += struct.pack(">I", len(encoded_id)) + encoded_id + salt return _h_a(material) + @staticmethod + def _entry_aad(entry: VaultEntry) -> bytes: + if ( + type(entry.format_version) is not int + or entry.format_version != 2 + or not isinstance(entry.entry_id, str) + or not entry.entry_id + or not isinstance(entry.nonce, bytes) + or len(entry.nonce) != 12 + or not isinstance(entry.salt, bytes) + or len(entry.salt) != 32 + or not isinstance(entry.tongue_affinity, str) + or not isinstance(entry.metadata, dict) + ): + raise ValueError("Unsupported or malformed authenticated vault entry") + for stamp in (entry.created_at, entry.expires_at): + if stamp is not None and (type(stamp) not in (int, float) or not math.isfinite(stamp)): + raise ValueError("Invalid vault time") + return json.dumps( + { + "domain": "scbe:braid-entry:v2", + "version": entry.format_version, + "entry_id": entry.entry_id, + "salt": entry.salt.hex(), + "created_at": entry.created_at, + "expires_at": entry.expires_at, + "tongue_affinity": entry.tongue_affinity, + "metadata": entry.metadata, + }, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("utf-8") + + def _seal_entry(self, entry_id, secret, created_at, expires_at, tongue_affinity, metadata, vault_key=None): + # Copy JSON metadata so later changes to the caller's dict cannot alter + # the stored authenticated header. Non-JSON or nonfinite data rejects. + metadata_copy = json.loads(json.dumps(metadata, allow_nan=False)) + entry = VaultEntry( + entry_id, b"", os.urandom(32), created_at, expires_at, tongue_affinity, metadata_copy, os.urandom(12) + ) + aad = self._entry_aad(entry) + key = self._derive_entry_key(entry_id, entry.salt, vault_key) + entry.ciphertext = AESGCM(key).encrypt(entry.nonce, secret, aad) + return entry + + def _decrypt_entry(self, entry_id: str, entry: VaultEntry) -> bytes: + try: + if entry.entry_id != entry_id: + raise ValueError("Entry identity mismatch") + aad = self._entry_aad(entry) + key = self._derive_entry_key(entry_id, entry.salt) + return AESGCM(key).decrypt(entry.nonce, entry.ciphertext, aad) + except (InvalidTag, ValueError, TypeError, OverflowError, AttributeError): + raise ValueError("Vault entry authentication failed") from None + def store( self, entry_id: str, @@ -336,28 +403,16 @@ def store( metadata: Optional[Dict[str, Any]] = None, ) -> VaultEntry: """Store a secret in the vault.""" - salt = os.urandom(32) - key = self._derive_entry_key(entry_id, salt) - - # Prepend 4-byte length header so we can strip padding on retrieval - payload = struct.pack(">I", len(secret)) + secret - - if len(payload) > len(key): - key = self._expand_key(key, len(payload)) - else: - payload = payload.ljust(len(key), b"\x00") - - ciphertext = _xor_bytes(payload, key) - + if ttl_seconds is not None and (type(ttl_seconds) not in (int, float) or not math.isfinite(ttl_seconds)): + raise ValueError("TTL must be finite") now = time.time() - entry = VaultEntry( - entry_id=entry_id, - ciphertext=ciphertext, - salt=salt, - created_at=now, - expires_at=now + ttl_seconds if ttl_seconds else None, - tongue_affinity=tongue_affinity, - metadata=metadata or {}, + entry = self._seal_entry( + entry_id, + secret, + now, + now + ttl_seconds if ttl_seconds is not None else None, + tongue_affinity, + {} if metadata is None else metadata, ) self._entries[entry_id] = entry self._log("store", entry_id) @@ -370,21 +425,14 @@ def retrieve(self, entry_id: str) -> Optional[bytes]: self._log("retrieve_miss", entry_id) return None + plaintext = self._decrypt_entry(entry_id, entry) if entry.is_expired: self._log("retrieve_expired", entry_id) del self._entries[entry_id] return None - key = self._derive_entry_key(entry_id, entry.salt) - if len(entry.ciphertext) > len(key): - key = self._expand_key(key, len(entry.ciphertext)) - - plaintext = _xor_bytes(entry.ciphertext, key) - # Extract original length from 4-byte header - secret_len = struct.unpack(">I", plaintext[:4])[0] - secret = plaintext[4 : 4 + secret_len] self._log("retrieve", entry_id) - return secret + return plaintext def rotate(self, entry_id: str, new_braid_key: BraidWord) -> Optional[VaultEntry]: """Rotate an entry's encryption under a new braid key. @@ -396,33 +444,36 @@ def rotate(self, entry_id: str, new_braid_key: BraidWord) -> Optional[VaultEntry if entry_id not in self._entries: return None - # Decrypt ALL entries with the current key + # Validate ALL records before changing the key or the entry map. decrypted = {} - for eid in list(self._entries): - val = self.retrieve(eid) - if val is not None: - decrypted[eid] = (val, self._entries[eid]) + for eid, entry in self._entries.items(): + val = self._decrypt_entry(eid, entry) + if not entry.is_expired: + decrypted[eid] = (val, entry) # Re-derive vault state with new braid key init = _init_strands(self._master_seed) new_strands = _apply_braid(init, new_braid_key) new_vault_key = _finalize(new_strands) - # Update vault state - self._braid_key = new_braid_key - self._derived_strands = new_strands - self._vault_key = new_vault_key - - # Re-encrypt ALL entries with the new key + # Stage fresh ciphertext, keeping original authenticated lifetime. + staged = {} for eid, (secret, old_entry) in decrypted.items(): - self.store( - entry_id=eid, - secret=secret, - ttl_seconds=(old_entry.expires_at - time.time() if old_entry.expires_at else None), - tongue_affinity=old_entry.tongue_affinity, - metadata=old_entry.metadata, + staged[eid] = self._seal_entry( + eid, + secret, + old_entry.created_at, + old_entry.expires_at, + old_entry.tongue_affinity, + old_entry.metadata, + new_vault_key, ) + self._braid_key = BraidWord(crossings=list(new_braid_key.crossings)) + self._derived_strands = new_strands + self._vault_key = new_vault_key + self._entries = staged + self._log("rotate", entry_id) return self._entries.get(entry_id) @@ -469,7 +520,11 @@ def _expand_key(self, key: bytes, length: int) -> bytes: return b"".join(blocks)[:length] def _purge_expired(self) -> None: - expired = [eid for eid, e in self._entries.items() if e.is_expired] + expired = [] + for eid, entry in self._entries.items(): + self._decrypt_entry(eid, entry) + if entry.is_expired: + expired.append(eid) for eid in expired: del self._entries[eid] self._log("auto_purge", eid) diff --git a/src/crypto/tri_bundle.py b/src/crypto/tri_bundle.py index d1a382404..c102d32ee 100644 --- a/src/crypto/tri_bundle.py +++ b/src/crypto/tri_bundle.py @@ -25,8 +25,9 @@ then the 3 bundles braid together (outer braid). Different compositions produce different cluster identities — non-commutative. -Scaling: (3^φ)³ ≈ 162 effective states per position. - With 6 tongues × 162 states = 972 dimensions per token. +Each cluster contains 27 numeric components; six tongue views contain 162. +The heuristic scaling factor (3^φ)^3 is about 207, not a count of independent +states, dimensions or entropy bits. Re-encoding a byte adds no source entropy. Author: SCBE-AETHERMOORE / Issac Davis """ @@ -79,7 +80,7 @@ ] # Scaling constant: (3^φ)³ -BUNDLE_SCALE = (3**PHI) ** 3 # ≈ 162.07 +BUNDLE_SCALE = (3**PHI) ** 3 # ≈ 207; heuristic, not an entropy estimate # --------------------------------------------------------------------------- @@ -211,7 +212,7 @@ def cluster_id_hex(self) -> str: @property def effective_states(self) -> float: - """(3^φ)³ ≈ 162 effective states per position.""" + """Legacy name for a heuristic scale, not a count of independent states.""" return BUNDLE_SCALE def energy(self) -> float: diff --git a/src/symphonic_cipher/scbe_aethermoore/spiral_seal/seal.py b/src/symphonic_cipher/scbe_aethermoore/spiral_seal/seal.py index 9f3ad951f..c8d7aff52 100644 --- a/src/symphonic_cipher/scbe_aethermoore/spiral_seal/seal.py +++ b/src/symphonic_cipher/scbe_aethermoore/spiral_seal/seal.py @@ -1,27 +1,10 @@ -""" -SpiralSeal SS1 - High-Level API -================================ -Post-quantum hybrid encryption using Kyber768 + Dilithium3 + AES-256-GCM. - -This is the main entry point for the 14-layer SCBE pipeline's cryptographic -operations. It provides: - -1. Hybrid encryption (Kyber768 KEM → AES-256-GCM) -2. Digital signatures (Dilithium3) -3. Sacred Tongue spell-text encoding -4. Key rotation support via key IDs (kid) - -Usage: - from spiral_seal import SpiralSealSS1 - - ss = SpiralSealSS1() - - # Encrypt - sealed = ss.seal(b"my secret API key", aad="service=openai;env=prod") - print(sealed) # SS1|kid=...|aad=...|salt=ru:...|nonce=ko:...|ct=ca:...|tag=dr:... +"""Legacy SS1 symmetric envelope API and detached signature helpers. - # Decrypt - plaintext = ss.unseal(sealed, aad="service=openai;env=prod") +The legacy SS1 serializer has no fields for a KEM ciphertext or signature. +Signed/hybrid envelope requests must therefore reject, never silently discard +security fields or imply verification. Use a separately reviewed protocol for +authenticated hybrid transport. This module is distinct from spiral_seal.py, +which supplies the package's exported SpiralSealSS1 class. """ from typing import Optional @@ -61,6 +44,8 @@ class SealedPayload: def to_ss1(self) -> str: """Format as SS1 spell-text blob.""" + if self.signature is not None or self.kyber_ct is not None: + raise ValueError("Legacy SS1 cannot serialize signature or KEM fields") return format_ss1_blob( kid=self.kid, aad=self.aad, @@ -75,7 +60,7 @@ class SpiralSealSS1: """ High-level API for SpiralSeal SS1 encryption. - Provides hybrid post-quantum encryption using: + Exposes symmetric envelopes and detached cryptographic helpers: - Kyber768 for key encapsulation - Dilithium3 for digital signatures - AES-256-GCM for symmetric encryption @@ -137,11 +122,13 @@ def seal(self, plaintext: bytes, aad: str = "", sign: bool = False) -> str: Args: plaintext: Data to encrypt aad: Additional authenticated data (e.g., "service=openai;env=prod") - sign: Whether to include a Dilithium signature (hybrid mode only) + sign: Must be False; this legacy format cannot carry signatures Returns: SS1 spell-text blob """ + if sign or self.mode != "symmetric": + raise ValueError("Legacy SS1 supports unsigned symmetric envelopes only") if isinstance(plaintext, str): plaintext = plaintext.encode("utf-8") @@ -192,7 +179,7 @@ def unseal(self, ss1_blob: str, aad: str = "", verify_sig: bool = False) -> byte Args: ss1_blob: SS1 spell-text blob aad: Additional authenticated data (must match seal) - verify_sig: Whether to verify Dilithium signature (hybrid mode) + verify_sig: Must be False; no signature exists in this legacy format Returns: Decrypted plaintext @@ -200,6 +187,8 @@ def unseal(self, ss1_blob: str, aad: str = "", verify_sig: bool = False) -> byte Raises: ValueError: If authentication fails or AAD mismatch """ + if verify_sig or self.mode != "symmetric": + raise ValueError("Legacy SS1 cannot verify a signed or hybrid envelope") # Parse the blob parsed = parse_ss1_blob(ss1_blob) diff --git a/src/symphonic_cipher/scbe_aethermoore/spiral_seal/signatures.py b/src/symphonic_cipher/scbe_aethermoore/spiral_seal/signatures.py index dd798adf8..77f292b7f 100644 --- a/src/symphonic_cipher/scbe_aethermoore/spiral_seal/signatures.py +++ b/src/symphonic_cipher/scbe_aethermoore/spiral_seal/signatures.py @@ -1,21 +1,31 @@ """ -Post-Quantum Digital Signatures - Dilithium3 -============================================= -Wrapper for Dilithium3 digital signature scheme. +ML-DSA-65 / legacy Dilithium3 signature wrapper. -Security level: ~AES-192 equivalent (NIST Level 3) -Collision probability: ~2^-128 - -Note: This module provides a fallback implementation using Ed25519 -when liboqs/pqcrypto is not available. For production, install the -official liboqs-python bindings. +Missing native support fails closed. Explicit isolated test settings may enable +real Ed25519 development signatures, which are labelled and NOT post-quantum. +Algorithm availability is not FIPS module validation. """ import os -import hashlib -import hmac from typing import Tuple +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey + +_SIG_ALG = "Dilithium3" +_DEV_PREFIX = b"DEV_ED25519:v1:" + + +def _development_enabled() -> bool: + return os.getenv("SCBE_ALLOW_MOCK_PQC") == "1" and os.getenv("SCBE_ENV", "").lower() in {"test", "development"} + + +def _require_development() -> None: + if not _development_enabled(): + raise RuntimeError("Native PQ signature backend unavailable; development signatures are disabled") + + # Try to import post-quantum library _FORCE_SKIP_LIBOQS = os.getenv("SCBE_FORCE_SKIP_LIBOQS", "").strip().lower() in { "1", @@ -77,11 +87,14 @@ def dilithium_keygen() -> Tuple[bytes, bytes]: return secret_key, public_key else: - # Fallback: HMAC-based simulation - # WARNING: This is NOT post-quantum secure! For development only. - secret_key = os.urandom(64) - public_key = hashlib.sha256(b"dilithium_pk_sim:" + secret_key[:32]).digest() - return secret_key, public_key + _require_development() + key = Ed25519PrivateKey.generate() + return ( + key.private_bytes( + serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption() + ), + key.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw), + ) def dilithium_sign(secret_key: bytes, message: bytes) -> bytes: @@ -105,11 +118,8 @@ def dilithium_sign(secret_key: bytes, message: bytes) -> bytes: return signature else: - # Fallback: HMAC-SHA256 simulation - # WARNING: This is NOT post-quantum secure! - signature = hmac.new(secret_key, message, hashlib.sha256).digest() - # Add a marker to identify fallback signatures - return b"FALLBACK_SIG:" + signature + _require_development() + return _DEV_PREFIX + Ed25519PrivateKey.from_private_bytes(secret_key).sign(message) def dilithium_verify(public_key: bytes, message: bytes, signature: bytes) -> bool: @@ -133,20 +143,18 @@ def dilithium_verify(public_key: bytes, message: bytes, signature: bytes) -> boo elif PQC_SIG_BACKEND == "pqcrypto": try: - dilithium.verify(public_key, message, signature) - return True + return dilithium.verify(public_key, message, signature) is True except Exception: return False else: - # Fallback verification - if not signature.startswith(b"FALLBACK_SIG:"): + if not _development_enabled() or not isinstance(signature, bytes) or not signature.startswith(_DEV_PREFIX): + return False + try: + Ed25519PublicKey.from_public_bytes(public_key).verify(signature[len(_DEV_PREFIX) :], message) + return True + except (InvalidSignature, ValueError, TypeError): return False - expected_sig = signature[13:] # Strip marker - # We need to derive the HMAC key from public key - # This is a simplified simulation - in reality we'd need the secret key - # For the fallback, we accept any signature that matches the format - return len(expected_sig) == 32 def get_pqc_sig_status() -> dict: @@ -159,9 +167,14 @@ def get_pqc_sig_status() -> dict: return { "available": PQC_SIG_AVAILABLE, "backend": PQC_SIG_BACKEND, - "algorithm": "Dilithium3", - "security_level": ("NIST Level 3 (~AES-192)" if PQC_SIG_AVAILABLE else "FALLBACK (NOT PQ-SECURE)"), + "algorithm": _SIG_ALG if PQC_SIG_AVAILABLE else "Ed25519-development", + "development_enabled": not PQC_SIG_AVAILABLE and _development_enabled(), + "security_level": ( + "Algorithm category 3; module validation not established" if PQC_SIG_AVAILABLE else "NOT PQ-SECURE" + ), "warning": ( - None if PQC_SIG_AVAILABLE else "Using classical fallback! Install liboqs-python for post-quantum security." + None + if PQC_SIG_AVAILABLE + else "Native PQ signatures unavailable; classical development mode requires explicit opt-in." ), } diff --git a/symphonic_cipher/scbe_aethermoore/spiral_seal/seal.py b/symphonic_cipher/scbe_aethermoore/spiral_seal/seal.py index 8255e0c4a..5fe5783ef 100644 --- a/symphonic_cipher/scbe_aethermoore/spiral_seal/seal.py +++ b/symphonic_cipher/scbe_aethermoore/spiral_seal/seal.py @@ -1,96 +1,64 @@ -""" -SpiralSeal SS1 - High-Level API -================================ -Post-quantum hybrid encryption using Kyber768 + Dilithium3 + AES-256-GCM. - -This is the main entry point for the 14-layer SCBE pipeline's cryptographic -operations. It provides: - -1. Hybrid encryption (Kyber768 KEM → AES-256-GCM) -2. Digital signatures (Dilithium3) -3. Sacred Tongue spell-text encoding -4. Key rotation support via key IDs (kid) - -Usage: - from spiral_seal import SpiralSealSS1 - - ss = SpiralSealSS1() - - # Encrypt - sealed = ss.seal(b"my secret API key", aad="service=openai;env=prod") - print(sealed) # SS1|kid=...|aad=...|salt=ru:...|nonce=ko:...|ct=ca:...|tag=dr:... - - # Decrypt - plaintext = ss.unseal(sealed, aad="service=openai;env=prod") +"""Legacy SS1 symmetric envelope API and detached signature helpers. + +The legacy SS1 serializer has no fields for a KEM ciphertext or signature. +Signed/hybrid envelope requests must therefore reject, never silently discard +security fields or imply verification. Use a separately reviewed protocol for +authenticated hybrid transport. This module is distinct from spiral_seal.py, +which supplies the package's exported SpiralSealSS1 class. """ -import os -import hashlib -from typing import Dict, Optional, Tuple +from typing import Optional from dataclasses import dataclass -from .key_exchange import kyber_keygen, kyber_encaps, kyber_decaps, get_pqc_status +from .key_exchange import kyber_keygen, kyber_encaps, get_pqc_status from .signatures import dilithium_keygen, dilithium_sign, dilithium_verify, get_pqc_sig_status -from .utils import ( - aes_gcm_encrypt, aes_gcm_decrypt, derive_key, - get_random, sha256, sha256_hex -) -from .sacred_tongues import ( - format_ss1_blob, parse_ss1_blob, - encode_to_spelltext, decode_from_spelltext -) +from .utils import aes_gcm_encrypt, aes_gcm_decrypt, derive_key, get_random +from .sacred_tongues import format_ss1_blob, parse_ss1_blob @dataclass class SealedPayload: """Container for a sealed (encrypted + signed) payload.""" - kid: str # Key ID for rotation - aad: str # Additional authenticated data - salt: bytes # KDF salt - nonce: bytes # AES-GCM nonce - ciphertext: bytes # Encrypted data - tag: bytes # AES-GCM auth tag - kyber_ct: Optional[bytes] # Kyber ciphertext (for hybrid mode) + + kid: str # Key ID for rotation + aad: str # Additional authenticated data + salt: bytes # KDF salt + nonce: bytes # AES-GCM nonce + ciphertext: bytes # Encrypted data + tag: bytes # AES-GCM auth tag + kyber_ct: Optional[bytes] # Kyber ciphertext (for hybrid mode) signature: Optional[bytes] # Dilithium signature - + def to_ss1(self) -> str: """Format as SS1 spell-text blob.""" + if self.signature is not None or self.kyber_ct is not None: + raise ValueError("Legacy SS1 cannot serialize signature or KEM fields") return format_ss1_blob( - kid=self.kid, - aad=self.aad, - salt=self.salt, - nonce=self.nonce, - ciphertext=self.ciphertext, - tag=self.tag + kid=self.kid, aad=self.aad, salt=self.salt, nonce=self.nonce, ciphertext=self.ciphertext, tag=self.tag ) class SpiralSealSS1: """ High-level API for SpiralSeal SS1 encryption. - - Provides hybrid post-quantum encryption using: + + Exposes symmetric envelopes and detached cryptographic helpers: - Kyber768 for key encapsulation - Dilithium3 for digital signatures - AES-256-GCM for symmetric encryption - Sacred Tongue tokenization for spell-text output - + Attributes: kid: Current key ID (for rotation) mode: 'hybrid' (Kyber+AES) or 'symmetric' (AES only) """ - - VERSION = 'SS1' - - def __init__( - self, - master_secret: Optional[bytes] = None, - kid: str = 'k01', - mode: str = 'symmetric' - ): + + VERSION = "SS1" + + def __init__(self, master_secret: Optional[bytes] = None, kid: str = "k01", mode: str = "symmetric"): """ Initialize SpiralSeal. - + Args: master_secret: 32-byte master secret for key derivation. If None, generates a random one (NOT for production!) @@ -99,76 +67,74 @@ def __init__( """ self.kid = kid self.mode = mode - + # Master secret (MUST be injected from env/KMS in production) if master_secret is None: import warnings + warnings.warn( "No master_secret provided - generating random key. " "This is NOT suitable for production! Inject via env/KMS.", - RuntimeWarning + RuntimeWarning, ) master_secret = get_random(32) - + if len(master_secret) != 32: raise ValueError("master_secret must be 32 bytes") - + self._master_secret = master_secret - + # Generate PQC keys if in hybrid mode - if mode == 'hybrid': + if mode == "hybrid": self._sk_enc, self._pk_enc = kyber_keygen() self._sk_sig, self._pk_sig = dilithium_keygen() else: self._sk_enc = self._pk_enc = None self._sk_sig = self._pk_sig = None - - def seal( - self, - plaintext: bytes, - aad: str = '', - sign: bool = False - ) -> str: + + def seal(self, plaintext: bytes, aad: str = "", sign: bool = False) -> str: """ Seal (encrypt) data and return SS1 spell-text. - + Args: plaintext: Data to encrypt aad: Additional authenticated data (e.g., "service=openai;env=prod") - sign: Whether to include a Dilithium signature (hybrid mode only) - + sign: Must be False; this legacy format cannot carry signatures + Returns: SS1 spell-text blob """ + if sign or self.mode != "symmetric": + raise ValueError("Legacy SS1 supports unsigned symmetric envelopes only") if isinstance(plaintext, str): - plaintext = plaintext.encode('utf-8') - + plaintext = plaintext.encode("utf-8") + # Generate salt for key derivation salt = get_random(16) - + # Derive encryption key - if self.mode == 'hybrid' and self._pk_enc: + if self.mode == "hybrid" and self._pk_enc: # Kyber key encapsulation kyber_ct, shared_secret = kyber_encaps(self._pk_enc) # Combine master secret with Kyber shared secret combined = self._master_secret + shared_secret - k_enc = derive_key(combined, salt, b'scbe:ss1:enc:v1') + k_enc = derive_key(combined, salt, b"scbe:ss1:enc:v1") else: # Symmetric-only mode kyber_ct = None - k_enc = derive_key(self._master_secret, salt, b'scbe:ss1:enc:v1') - + k_enc = derive_key(self._master_secret, salt, b"scbe:ss1:enc:v1") + # Encrypt with AES-256-GCM - aad_bytes = aad.encode('utf-8') if aad else b'' + aad_bytes = aad.encode("utf-8") if aad else b"" nonce, ciphertext, tag = aes_gcm_encrypt(k_enc, plaintext, aad_bytes) - + # Optional signature signature = None - if sign and self.mode == 'hybrid' and self._sk_sig: + if sign and self.mode == "hybrid" and self._sk_sig: # Sign the ciphertext + AAD to_sign = ciphertext + tag + aad_bytes signature = dilithium_sign(self._sk_sig, to_sign) - + # Create payload payload = SealedPayload( kid=self.kid, @@ -178,124 +144,121 @@ def seal( ciphertext=ciphertext, tag=tag, kyber_ct=kyber_ct, - signature=signature + signature=signature, ) - + return payload.to_ss1() - - def unseal( - self, - ss1_blob: str, - aad: str = '', - verify_sig: bool = False - ) -> bytes: + + def unseal(self, ss1_blob: str, aad: str = "", verify_sig: bool = False) -> bytes: """ Unseal (decrypt) an SS1 spell-text blob. - + Args: ss1_blob: SS1 spell-text blob aad: Additional authenticated data (must match seal) - verify_sig: Whether to verify Dilithium signature (hybrid mode) - + verify_sig: Must be False; no signature exists in this legacy format + Returns: Decrypted plaintext - + Raises: ValueError: If authentication fails or AAD mismatch """ + if verify_sig or self.mode != "symmetric": + raise ValueError("Legacy SS1 cannot verify a signed or hybrid envelope") # Parse the blob parsed = parse_ss1_blob(ss1_blob) - + # Verify AAD matches - if parsed.get('aad', '') != aad: + if parsed.get("aad", "") != aad: raise ValueError("AAD mismatch - authentication failed") - + # Extract components - salt = parsed['salt'] - nonce = parsed['nonce'] - ciphertext = parsed['ct'] - tag = parsed['tag'] - + salt = parsed["salt"] + nonce = parsed["nonce"] + ciphertext = parsed["ct"] + tag = parsed["tag"] + # Derive decryption key - if self.mode == 'hybrid' and self._sk_enc: + if self.mode == "hybrid" and self._sk_enc: # In a real implementation, kyber_ct would be in the blob # For now, we use symmetric derivation - k_enc = derive_key(self._master_secret, salt, b'scbe:ss1:enc:v1') + k_enc = derive_key(self._master_secret, salt, b"scbe:ss1:enc:v1") else: - k_enc = derive_key(self._master_secret, salt, b'scbe:ss1:enc:v1') - + k_enc = derive_key(self._master_secret, salt, b"scbe:ss1:enc:v1") + # Decrypt - aad_bytes = aad.encode('utf-8') if aad else b'' + aad_bytes = aad.encode("utf-8") if aad else b"" try: plaintext = aes_gcm_decrypt(k_enc, nonce, ciphertext, tag, aad_bytes) return plaintext - except ValueError as e: + except ValueError: # Fail-to-noise: don't expose details raise ValueError("Authentication failed") from None - + def sign(self, message: bytes) -> bytes: """ Sign a message using Dilithium3. - + Args: message: Message to sign - + Returns: Digital signature - + Raises: RuntimeError: If not in hybrid mode """ - if self.mode != 'hybrid' or not self._sk_sig: + if self.mode != "hybrid" or not self._sk_sig: raise RuntimeError("Signing requires hybrid mode") return dilithium_sign(self._sk_sig, message) - + def verify(self, message: bytes, signature: bytes) -> bool: """ Verify a Dilithium3 signature. - + Args: message: Original message signature: Signature to verify - + Returns: True if valid, False otherwise """ - if self.mode != 'hybrid' or not self._pk_sig: + if self.mode != "hybrid" or not self._pk_sig: raise RuntimeError("Verification requires hybrid mode") return dilithium_verify(self._pk_sig, message, signature) - + def rotate_key(self, new_kid: str, new_master_secret: bytes): """ Rotate to a new master secret. - + Args: new_kid: New key ID new_master_secret: New 32-byte master secret """ if len(new_master_secret) != 32: raise ValueError("master_secret must be 32 bytes") - + self.kid = new_kid self._master_secret = new_master_secret - + # Regenerate PQC keys if in hybrid mode - if self.mode == 'hybrid': + if self.mode == "hybrid": self._sk_enc, self._pk_enc = kyber_keygen() self._sk_sig, self._pk_sig = dilithium_keygen() - + @staticmethod def get_status() -> dict: """ Get the status of cryptographic backends. - + Returns: Dict with PQC availability and security info """ return { - 'version': SpiralSealSS1.VERSION, - 'key_exchange': get_pqc_status(), - 'signatures': get_pqc_sig_status(), + "version": SpiralSealSS1.VERSION, + "key_exchange": get_pqc_status(), + "signatures": get_pqc_sig_status(), } @@ -303,43 +266,35 @@ def get_status() -> dict: # CONVENIENCE FUNCTIONS # ============================================================================= -def seal( - plaintext: bytes, - master_secret: bytes, - aad: str = '', - kid: str = 'k01' -) -> str: + +def seal(plaintext: bytes, master_secret: bytes, aad: str = "", kid: str = "k01") -> str: """ One-shot seal function. - + Args: plaintext: Data to encrypt master_secret: 32-byte master secret aad: Additional authenticated data kid: Key ID - + Returns: SS1 spell-text blob """ - ss = SpiralSealSS1(master_secret=master_secret, kid=kid, mode='symmetric') + ss = SpiralSealSS1(master_secret=master_secret, kid=kid, mode="symmetric") return ss.seal(plaintext, aad=aad) -def unseal( - ss1_blob: str, - master_secret: bytes, - aad: str = '' -) -> bytes: +def unseal(ss1_blob: str, master_secret: bytes, aad: str = "") -> bytes: """ One-shot unseal function. - + Args: ss1_blob: SS1 spell-text blob master_secret: 32-byte master secret aad: Additional authenticated data (must match seal) - + Returns: Decrypted plaintext """ - ss = SpiralSealSS1(master_secret=master_secret, mode='symmetric') + ss = SpiralSealSS1(master_secret=master_secret, mode="symmetric") return ss.unseal(ss1_blob, aad=aad) diff --git a/symphonic_cipher/scbe_aethermoore/spiral_seal/signatures.py b/symphonic_cipher/scbe_aethermoore/spiral_seal/signatures.py index b6ecefdec..77f292b7f 100644 --- a/symphonic_cipher/scbe_aethermoore/spiral_seal/signatures.py +++ b/symphonic_cipher/scbe_aethermoore/spiral_seal/signatures.py @@ -1,21 +1,31 @@ """ -Post-Quantum Digital Signatures - Dilithium3 -============================================= -Wrapper for Dilithium3 digital signature scheme. +ML-DSA-65 / legacy Dilithium3 signature wrapper. -Security level: ~AES-192 equivalent (NIST Level 3) -Collision probability: ~2^-128 - -Note: This module provides a fallback implementation using Ed25519 -when liboqs/pqcrypto is not available. For production, install the -official liboqs-python bindings. +Missing native support fails closed. Explicit isolated test settings may enable +real Ed25519 development signatures, which are labelled and NOT post-quantum. +Algorithm availability is not FIPS module validation. """ import os -import hashlib -import hmac from typing import Tuple +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey + +_SIG_ALG = "Dilithium3" +_DEV_PREFIX = b"DEV_ED25519:v1:" + + +def _development_enabled() -> bool: + return os.getenv("SCBE_ALLOW_MOCK_PQC") == "1" and os.getenv("SCBE_ENV", "").lower() in {"test", "development"} + + +def _require_development() -> None: + if not _development_enabled(): + raise RuntimeError("Native PQ signature backend unavailable; development signatures are disabled") + + # Try to import post-quantum library _FORCE_SKIP_LIBOQS = os.getenv("SCBE_FORCE_SKIP_LIBOQS", "").strip().lower() in { "1", @@ -27,30 +37,34 @@ try: from oqs import Signature import oqs as _oqs_mod + PQC_SIG_AVAILABLE = True - PQC_SIG_BACKEND = 'liboqs' + PQC_SIG_BACKEND = "liboqs" + # Select algorithm: prefer ML-DSA-65 (FIPS 204), fall back to Dilithium3 _enabled = _oqs_mod.get_enabled_sig_mechanisms() _SIG_ALG = "ML-DSA-65" if "ML-DSA-65" in _enabled else "Dilithium3" except BaseException: + # liboqs-python not installed, shared libs missing, or bootstrap errors try: import pqcrypto.sign.dilithium3 as dilithium + PQC_SIG_AVAILABLE = True - PQC_SIG_BACKEND = 'pqcrypto' + PQC_SIG_BACKEND = "pqcrypto" except BaseException: PQC_SIG_AVAILABLE = False - PQC_SIG_BACKEND = 'fallback' + PQC_SIG_BACKEND = "fallback" else: PQC_SIG_AVAILABLE = False - PQC_SIG_BACKEND = 'fallback' + PQC_SIG_BACKEND = "fallback" class DilithiumKeyPair: """Container for Dilithium3 key pair.""" - + def __init__(self, public_key: bytes, secret_key: bytes): self.public_key = public_key self.secret_key = secret_key - + def __repr__(self): return f"DilithiumKeyPair(pk={len(self.public_key)}B, sk={len(self.secret_key)}B)" @@ -58,105 +72,109 @@ def __repr__(self): def dilithium_keygen() -> Tuple[bytes, bytes]: """ Generate a Dilithium3 key pair. - + Returns: Tuple of (secret_key, public_key) """ - if PQC_SIG_BACKEND == 'liboqs': + if PQC_SIG_BACKEND == "liboqs": sig = Signature(_SIG_ALG) public_key = sig.generate_keypair() secret_key = sig.export_secret_key() return secret_key, public_key - - elif PQC_SIG_BACKEND == 'pqcrypto': + + elif PQC_SIG_BACKEND == "pqcrypto": public_key, secret_key = dilithium.generate_keypair() return secret_key, public_key - + else: - # Fallback: HMAC-based simulation - # WARNING: This is NOT post-quantum secure! For development only. - secret_key = os.urandom(64) - public_key = hashlib.sha256(b'dilithium_pk_sim:' + secret_key[:32]).digest() - return secret_key, public_key + _require_development() + key = Ed25519PrivateKey.generate() + return ( + key.private_bytes( + serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption() + ), + key.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw), + ) def dilithium_sign(secret_key: bytes, message: bytes) -> bytes: """ Sign a message using Dilithium3. - + Args: secret_key: Signer's secret key message: Message to sign - + Returns: Digital signature """ - if PQC_SIG_BACKEND == 'liboqs': + if PQC_SIG_BACKEND == "liboqs": sig = Signature(_SIG_ALG, secret_key) signature = sig.sign(message) return signature - - elif PQC_SIG_BACKEND == 'pqcrypto': + + elif PQC_SIG_BACKEND == "pqcrypto": signature = dilithium.sign(secret_key, message) return signature - + else: - # Fallback: HMAC-SHA256 simulation - # WARNING: This is NOT post-quantum secure! - signature = hmac.new(secret_key, message, hashlib.sha256).digest() - # Add a marker to identify fallback signatures - return b'FALLBACK_SIG:' + signature + _require_development() + return _DEV_PREFIX + Ed25519PrivateKey.from_private_bytes(secret_key).sign(message) def dilithium_verify(public_key: bytes, message: bytes, signature: bytes) -> bool: """ Verify a Dilithium3 signature. - + Args: public_key: Signer's public key message: Original message signature: Signature to verify - + Returns: True if signature is valid, False otherwise """ - if PQC_SIG_BACKEND == 'liboqs': + if PQC_SIG_BACKEND == "liboqs": sig = Signature(_SIG_ALG) try: return sig.verify(message, signature, public_key) except Exception: return False - - elif PQC_SIG_BACKEND == 'pqcrypto': + + elif PQC_SIG_BACKEND == "pqcrypto": try: - dilithium.verify(public_key, message, signature) - return True + return dilithium.verify(public_key, message, signature) is True except Exception: return False - + else: - # Fallback verification - if not signature.startswith(b'FALLBACK_SIG:'): + if not _development_enabled() or not isinstance(signature, bytes) or not signature.startswith(_DEV_PREFIX): + return False + try: + Ed25519PublicKey.from_public_bytes(public_key).verify(signature[len(_DEV_PREFIX) :], message) + return True + except (InvalidSignature, ValueError, TypeError): return False - expected_sig = signature[13:] # Strip marker - # We need to derive the HMAC key from public key - # This is a simplified simulation - in reality we'd need the secret key - # For the fallback, we accept any signature that matches the format - return len(expected_sig) == 32 def get_pqc_sig_status() -> dict: """ Get the status of post-quantum signature support. - + Returns: Dict with backend info and security warnings """ return { - 'available': PQC_SIG_AVAILABLE, - 'backend': PQC_SIG_BACKEND, - 'algorithm': 'Dilithium3', - 'security_level': 'NIST Level 3 (~AES-192)' if PQC_SIG_AVAILABLE else 'FALLBACK (NOT PQ-SECURE)', - 'warning': None if PQC_SIG_AVAILABLE else - 'Using classical fallback! Install liboqs-python for post-quantum security.' + "available": PQC_SIG_AVAILABLE, + "backend": PQC_SIG_BACKEND, + "algorithm": _SIG_ALG if PQC_SIG_AVAILABLE else "Ed25519-development", + "development_enabled": not PQC_SIG_AVAILABLE and _development_enabled(), + "security_level": ( + "Algorithm category 3; module validation not established" if PQC_SIG_AVAILABLE else "NOT PQ-SECURE" + ), + "warning": ( + None + if PQC_SIG_AVAILABLE + else "Native PQ signatures unavailable; classical development mode requires explicit opt-in." + ), } diff --git a/tests/security/test_braid_signature_integrity.py b/tests/security/test_braid_signature_integrity.py new file mode 100644 index 000000000..c7daecb47 --- /dev/null +++ b/tests/security/test_braid_signature_integrity.py @@ -0,0 +1,207 @@ +"""Tamper and downgrade boundaries for custom composition and legacy signing.""" + +import importlib.util +import sys +import types +from pathlib import Path + +import pytest + +from src.crypto.braid_vault import BraidVault, BraidWord + +ROOT = Path(__file__).resolve().parents[2] + + +@pytest.fixture(params=["src/symphonic_cipher", "symphonic_cipher"]) +def signatures(request, monkeypatch): + monkeypatch.setenv("SCBE_FORCE_SKIP_LIBOQS", "1") + monkeypatch.delenv("SCBE_ALLOW_MOCK_PQC", raising=False) + monkeypatch.delenv("SCBE_ENV", raising=False) + path = ROOT / request.param / "scbe_aethermoore/spiral_seal/signatures.py" + spec = importlib.util.spec_from_file_location("review_sig_" + request.param.replace("/", "_"), path) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def test_signature_format_alone_never_authenticates(signatures): + assert not signatures.dilithium_verify(b"not-a-key", b"forged", b"FALLBACK_SIG:" + b"x" * 32) + + +def test_missing_native_signer_fails_closed(signatures): + with pytest.raises(RuntimeError): + signatures.dilithium_keygen() + with pytest.raises(RuntimeError): + signatures.dilithium_sign(b"k" * 32, b"message") + + +def test_explicit_development_signatures_bind_message_and_key(signatures, monkeypatch): + monkeypatch.setenv("SCBE_ALLOW_MOCK_PQC", "1") + monkeypatch.setenv("SCBE_ENV", "test") + sk, pk = signatures.dilithium_keygen() + _, wrong_pk = signatures.dilithium_keygen() + signature = signatures.dilithium_sign(sk, b"message") + assert signatures.dilithium_verify(pk, b"message", signature) + assert not signatures.dilithium_verify(pk, b"changed", signature) + assert not signatures.dilithium_verify(wrong_pk, b"message", signature) + assert not signatures.dilithium_verify(pk, b"message", signature[:-1]) + assert signatures.get_pqc_sig_status()["algorithm"] == "Ed25519-development" + monkeypatch.delenv("SCBE_ALLOW_MOCK_PQC") + assert not signatures.dilithium_verify(pk, b"message", signature) + + +def test_production_cannot_enable_development_signer(signatures, monkeypatch): + monkeypatch.setenv("SCBE_ALLOW_MOCK_PQC", "1") + monkeypatch.setenv("SCBE_ENV", "production") + with pytest.raises(RuntimeError): + signatures.dilithium_keygen() + + +@pytest.fixture +def vault(): + return BraidVault(b"public-test-seed" * 3, BraidWord.decode("s1")) + + +@pytest.mark.parametrize( + "part", + [ + "ciphertext", + "salt", + "nonce", + "metadata", + "entry_id", + "tongue_affinity", + "expires_at", + "created_at", + "format_version", + ], +) +def test_vault_rejects_altered_entry(vault, part): + entry = vault.store("demo", b"public test content", metadata={"scope": "test"}) + if part in {"ciphertext", "salt", "nonce"}: + data = bytearray(getattr(entry, part)) + data[4] ^= 1 + setattr(entry, part, bytes(data)) + elif part == "metadata": + entry.metadata["scope"] = "other" + elif part in {"expires_at", "created_at"}: + setattr(entry, part, 1.0) + elif part == "format_version": + entry.format_version = 1 + else: + setattr(entry, part, "different") + with pytest.raises(ValueError): + vault.retrieve("demo") + + +def test_rotation_validates_all_entries_before_committing(vault): + vault.store("a", b"first") + entry = vault.store("b", b"second") + entry.ciphertext = entry.ciphertext[:-1] + bytes([entry.ciphertext[-1] ^ 1]) + old_key, old_cipher = vault._vault_key, vault._entries["a"].ciphertext + with pytest.raises(ValueError): + vault.rotate("a", BraidWord.decode("s2")) + assert vault._vault_key == old_key + assert vault._entries["a"].ciphertext == old_cipher + assert vault.retrieve("a") == b"first" + + +def test_zero_ttl_is_expired_not_immortal(vault): + vault.store("instant", b"test", ttl_seconds=0) + assert vault.retrieve("instant") is None + + +@pytest.mark.parametrize("result", [False, None, True]) +def test_pqcrypto_boolean_verdict_is_respected(signatures, monkeypatch, result): + # Contract test only; real native validation runs separately in CI. + monkeypatch.setattr(signatures, "PQC_SIG_BACKEND", "pqcrypto") + monkeypatch.setattr(signatures, "dilithium", types.SimpleNamespace(verify=lambda *args: result), raising=False) + assert signatures.dilithium_verify(b"key", b"msg", b"sig") is (result is True) + + +@pytest.fixture(params=["src/symphonic_cipher", "symphonic_cipher"]) +def legacy_seal(request, monkeypatch): + monkeypatch.setenv("SCBE_FORCE_SKIP_LIBOQS", "1") + name = "review_seal_" + request.param.replace("/", "_") + directory = ROOT / request.param / "scbe_aethermoore/spiral_seal" + package = types.ModuleType(name) + package.__path__ = [str(directory)] + monkeypatch.setitem(sys.modules, name, package) + return __import__(name + ".seal", fromlist=["seal"]) + + +def test_legacy_unsigned_roundtrip_and_aad_binding(legacy_seal): + api = legacy_seal.SpiralSealSS1(master_secret=b"x" * 32) + blob = api.seal(b"test", aad="purpose=test") + assert api.unseal(blob, aad="purpose=test") == b"test" + with pytest.raises(ValueError): + api.unseal(blob, aad="other") + + +def test_legacy_required_signature_cannot_be_ignored(legacy_seal): + api = legacy_seal.SpiralSealSS1(master_secret=b"x" * 32) + blob = api.seal(b"test") + with pytest.raises(ValueError): + api.unseal(blob, verify_sig=True) + with pytest.raises(ValueError): + api.seal(b"test", sign=True) + + +@pytest.mark.parametrize("field", ["signature", "kyber_ct"]) +def test_legacy_serializer_never_discards_security_fields(legacy_seal, field): + payload = legacy_seal.SealedPayload("k01", "", b"s" * 16, b"n" * 12, b"ct", b"tag", None, None) + setattr(payload, field, b"supplied security data") + with pytest.raises(ValueError): + payload.to_ss1() + + +def test_vault_metadata_copy_and_rotation_preserve_lifetime(vault): + metadata = {"nested": {"scope": "demo"}} + entry = vault.store("a", b"first", ttl_seconds=300, metadata=metadata) + vault.store("b", b"second") + metadata["nested"]["scope"] = "changed" + assert vault.retrieve("a") == b"first" + rotated = vault.rotate("a", BraidWord.decode("s2")) + assert rotated.created_at == entry.created_at + assert rotated.expires_at == entry.expires_at + assert rotated.nonce != entry.nonce + assert rotated.metadata == {"nested": {"scope": "demo"}} + assert vault.retrieve("a") == b"first" + assert vault.retrieve("b") == b"second" + + +@pytest.mark.parametrize("attack", ["truncated", "wrong_master", "swapped_id", "legacy"]) +def test_vault_alternate_tampering(vault, attack): + entry = vault.store("a", b"first") + if attack == "truncated": + entry.ciphertext = entry.ciphertext[:8] + elif attack == "wrong_master": + vault._vault_key = b"w" * 32 + elif attack == "legacy": + entry.nonce = b"" + else: + entry.entry_id = "b" + vault._entries["b"] = entry + with pytest.raises(ValueError): + vault.retrieve("b" if attack == "swapped_id" else "a") + + +def test_rotation_staging_failure_leaves_old_state(vault, monkeypatch): + vault.store("a", b"first") + vault.store("b", b"second") + old_key, old_entries = vault._vault_key, vault._entries + original = vault._seal_entry + + def fail_second(entry_id, *args): + if entry_id == "b": + raise RuntimeError("simulated encryption failure") + return original(entry_id, *args) + + monkeypatch.setattr(vault, "_seal_entry", fail_second) + with pytest.raises(RuntimeError): + vault.rotate("a", BraidWord.decode("s2")) + assert vault._vault_key == old_key + assert vault._entries is old_entries + assert vault.retrieve("a") == b"first" + assert vault.retrieve("b") == b"second" diff --git a/tests/test_braid_vault.py b/tests/test_braid_vault.py index 5266d9aee..92d20a152 100644 --- a/tests/test_braid_vault.py +++ b/tests/test_braid_vault.py @@ -1,8 +1,8 @@ -"""Tests for the Braided Dual-Primitive Key Vault. +"""Tests for the Braid-Inspired Key Vault. Covers: braid word operations, dual hashing, strand initialization, crossing mechanics, vault CRUD, key rotation, TTL expiry, and -the non-commutativity property that gives topological security. +order sensitivity (not a cryptographic hardness proof). """ from __future__ import annotations @@ -39,7 +39,7 @@ def test_h_b_produces_32_bytes(self): assert len(_h_b(b"test")) == 32 def test_h_a_and_h_b_differ(self): - """SHA3-256 and BLAKE2b must produce different digests.""" + """Distinct PBKDF2 domains produce different digests for this input.""" assert _h_a(b"same input") != _h_b(b"same input") def test_dual_hash_returns_pair(self): diff --git a/tests/test_industry_grade.py b/tests/test_industry_grade.py index bd1f3f941..742a4d499 100644 --- a/tests/test_industry_grade.py +++ b/tests/test_industry_grade.py @@ -37,6 +37,14 @@ import pytest + +@pytest.fixture(autouse=True) +def explicit_development_signatures(monkeypatch): + """Legacy interface checks may use the explicitly labelled classical signer.""" + monkeypatch.setenv("SCBE_ALLOW_MOCK_PQC", "1") + monkeypatch.setenv("SCBE_ENV", "test") + + try: from cryptography.fernet import Fernet @@ -1411,10 +1419,7 @@ def test_164_dilithium_wrong_public_key(self): sig = dilithium_sign(sk1, b"test") - # Verification with wrong key (behavior depends on backend) - result = dilithium_verify(pk2, b"test", sig) - # In real PQC, this would be False - assert isinstance(result, bool) + assert not dilithium_verify(pk2, b"test", sig) def test_165_pqc_status_reporting(self): """PQC status should report algorithm details.""" @@ -1422,7 +1427,7 @@ def test_165_pqc_status_reporting(self): sig_status = get_pqc_sig_status() assert ke_status["algorithm"] == "Kyber768" - assert sig_status["algorithm"] == "Dilithium3" + assert sig_status["algorithm"] in {"ML-DSA-65", "Dilithium3", "Ed25519-development"} assert "backend" in ke_status assert "backend" in sig_status diff --git a/tests/test_scbe_comprehensive.py b/tests/test_scbe_comprehensive.py index 602589798..2d50307bc 100644 --- a/tests/test_scbe_comprehensive.py +++ b/tests/test_scbe_comprehensive.py @@ -26,6 +26,14 @@ import pytest + +@pytest.fixture(autouse=True) +def explicit_development_signatures(monkeypatch): + """Legacy interface checks may use the explicitly labelled classical signer.""" + monkeypatch.setenv("SCBE_ALLOW_MOCK_PQC", "1") + monkeypatch.setenv("SCBE_ENV", "test") + + try: from cryptography import fernet @@ -781,12 +789,7 @@ def test_70_dilithium_wrong_message_fails(self): """Dilithium3 verify should fail for wrong message.""" sk, pk = dilithium_keygen() signature = dilithium_sign(sk, b"original") - # Fallback mode accepts format, real PQC would fail - # This tests the interface at minimum - result = dilithium_verify(pk, b"tampered", signature) - # In fallback mode, this may pass (format check only) - # In real PQC mode, this would fail - assert isinstance(result, bool) + assert not dilithium_verify(pk, b"tampered", signature) def test_71_dilithium_signature_not_empty(self): """Dilithium3 signature should not be empty.""" @@ -799,7 +802,7 @@ def test_72_dilithium_status(self): status = get_pqc_sig_status() assert "backend" in status assert "algorithm" in status - assert status["algorithm"] == "Dilithium3" + assert status["algorithm"] in {"ML-DSA-65", "Dilithium3", "Ed25519-development"} def test_73_pqc_fallback_warning(self): """PQC status should warn if using fallback.""" diff --git a/tests/test_spiral_seal_comprehensive.py b/tests/test_spiral_seal_comprehensive.py index 80137cf1a..362dd8599 100644 --- a/tests/test_spiral_seal_comprehensive.py +++ b/tests/test_spiral_seal_comprehensive.py @@ -21,6 +21,14 @@ import pytest + +@pytest.fixture(autouse=True) +def explicit_development_signatures(monkeypatch): + """Legacy interface checks may use the explicitly labelled classical signer.""" + monkeypatch.setenv("SCBE_ALLOW_MOCK_PQC", "1") + monkeypatch.setenv("SCBE_ENV", "test") + + try: from cryptography import fernet @@ -687,31 +695,17 @@ def test_57_dilithium_sign_verify_roundtrip(self): # Test 58: Dilithium wrong message fails verification def test_58_dilithium_wrong_message_fails(self): """Wrong message should fail Dilithium verification (when PQC available).""" - from symphonic_cipher.scbe_aethermoore.spiral_seal.signatures import ( - PQC_SIG_AVAILABLE, - ) - sk, pk = dilithium_keygen() message = b"original message" signature = dilithium_sign(sk, message) is_valid = dilithium_verify(pk, b"wrong message", signature) - # Note: Fallback HMAC-based verification is simplified and may pass - # Real PQC implementation would fail here - if PQC_SIG_AVAILABLE: - assert not is_valid, "Wrong message passed verification" - else: - # Fallback mode - just verify the function runs - assert isinstance(is_valid, bool) + assert not is_valid, "Wrong message passed verification" # Test 59: Dilithium wrong key fails verification def test_59_dilithium_wrong_key_fails(self): """Wrong public key should fail Dilithium verification (when PQC available).""" - from symphonic_cipher.scbe_aethermoore.spiral_seal.signatures import ( - PQC_SIG_AVAILABLE, - ) - sk1, pk1 = dilithium_keygen() sk2, pk2 = dilithium_keygen() @@ -720,11 +714,7 @@ def test_59_dilithium_wrong_key_fails(self): is_valid = dilithium_verify(pk2, message, signature) - # Note: Fallback mode has simplified verification - if PQC_SIG_AVAILABLE: - assert not is_valid, "Wrong key passed verification" - else: - assert isinstance(is_valid, bool) + assert not is_valid, "Wrong key passed verification" # Test 60: Dilithium signature is non-empty def test_60_dilithium_signature_non_empty(self): From 67a1e38b189de3550f17571ca912d4a237973917 Mon Sep 17 00:00:00 2001 From: issdandavis <215328633+issdandavis@users.noreply.github.com> Date: Fri, 18 Sep 2026 17:59:09 -0700 Subject: [PATCH 2/3] fix(crypto): bind RWP2 authentication and reject demo verification --- .../HASH_AND_SIGNATURE_REVIEW_2026-09-18.md | 57 +++++- scripts/system/run_core_python_checks.py | 1 + src/crypto/sacred_eggs.py | 4 + src/spiralverse/aethercode.py | 65 +++++-- src/spiralverse/rwp2_envelope.py | 128 +++++++++----- .../test_rwp2_authentication_contract.py | 165 ++++++++++++++++++ tests/test_full_system_verification.py | 4 +- 7 files changed, 357 insertions(+), 67 deletions(-) create mode 100644 tests/security/test_rwp2_authentication_contract.py diff --git a/docs/security/HASH_AND_SIGNATURE_REVIEW_2026-09-18.md b/docs/security/HASH_AND_SIGNATURE_REVIEW_2026-09-18.md index ff90d8a63..7d6c8a123 100644 --- a/docs/security/HASH_AND_SIGNATURE_REVIEW_2026-09-18.md +++ b/docs/security/HASH_AND_SIGNATURE_REVIEW_2026-09-18.md @@ -22,7 +22,7 @@ around them. That can be useful without claiming a new secure hash primitive. | `src/crypto/sacred_eggs.py` | SHA-256 shells, HMAC/HKDF-SHA256 and a sorted three-shell binding | Commitments, derivation and contextual binding. Three shells do not imply triple security. Showing a secret to a verifier is not a zero-knowledge proof. | | `packages/kernel/src/tBraid.ts` | Derived temporal variants, weighted hyperbolic distances and an exponential score | A routing/geometry measure, not a cryptographic hash or signature. | | `python/scbe/elastic_bijective_hash.py` | Reversible SplitMix64-based mapping with a double-hashed lookup table | Indexing and memory lookup. Not a collision-resistant security primitive. | -| `src/spiralverse/rwp2_envelope.py` | Per-tongue HMAC-SHA256 | Shared-secret authentication, not public-key signatures. Further blockers are listed below. | +| `src/spiralverse/rwp2_envelope.py` | Per-tongue, domain-separated HMAC-SHA256 with explicit keys and a canonical envelope | Shared-secret authentication, not public-key signatures. Freshness state is process-local. | | Both `spiral_seal/signatures.py` copies | Native liboqs ML-DSA-65 when available, or legacy Dilithium3; explicitly enabled Ed25519 for development | Actual signing. Ed25519 development mode is classical, not PQ-secure. A standard algorithm alone does not establish FIPS module validation. | ## Confirmed signing and vault defects repaired @@ -64,14 +64,15 @@ encryption does not prove that derivation secure or repair low-entropy secrets. ## Remaining blockers and limits -- RWP2 currently defaults to deterministic, source-visible tongue keys. Its - `|`-joined MAC input is ambiguous for fields containing delimiters, and omits - `kid`, `tier` and `version`. Empty required-tongue sets can pass vacuously. - Do not deploy this legacy path as an authentication boundary until the full - key provisioning and canonical-input contract is repaired and tested. -- Aethercode's `LEDGER SIGN` is a truncated SHA-256 composition with a demo key, - not HMAC or public-key signing. Its execution-proof caller also uses the legacy - RWP2 defaults. Interpreter output must not be treated as security proof. +- Custom derivation and braid constructions still need independent + cryptanalysis. Passing functional/tamper tests cannot establish hardness. +- RWP2 replay state is per process and does not survive restarts. Deployments + with multiple workers need a shared atomic freshness store. Key provisioning, + rotation, custody and policy selection remain operator responsibilities. + Possessing multiple tongue keys does not establish multiple independent actors. +- Aethercode's authenticated receipt only authenticates the supplied trace. + It does not prove execution or truth. Its `PROOF` command still builds a + descriptive trace summary, not a formal or cryptographic computation proof. - `create_session_egg` derives its yolk from the session ID. If that ID is public, the yolk is public too. It is suitable as a deterministic identifier, not an independently secret authentication credential. @@ -83,6 +84,44 @@ encryption does not prove that derivation secure or repair low-entropy secrets. do not create fresh secret entropy. Geometry may route a decision but cannot authorize a missing or invalid signature. +## Connected RWP2 and Aethercode repair + +Follow-up inspection reproduced eight concrete bad outcomes on the pre-repair +commit: authentication with public demo keys; accepted changes to key ID, tier +and version; accepted delimiter redistribution; accepted empty required signer +sets; replay after eviction of a live cache record; and `LEDGER VERIFY nonsense` +returning true. The corrected contract is: + +- `SignatureEngine(keys=...)` and `EnvelopeFactory(keys=...)` require explicitly + supplied, nonempty keys. Source-visible historical demo keys are rejected. + Keys are copied so later caller dictionary changes cannot switch the verifier. +- MAC revision 2 encodes typed fields as canonical JSON, binary payload as + Base64URL, and authenticates every field including `kid`, `tier` and `version`. + Each tongue has its own domain label. Old delimiter-based tags are rejected. +- Every tier's minimum signers must verify. An explicit required set may add + requirements, but cannot remove the tier minimum or be empty. Missing signing + keys raise instead of silently creating a partial result. +- The factory verifies the MAC before consuming a nonce. A lock makes check and + record atomic within one process. A full replay cache rejects new work instead + of evicting unexpired receipts and allowing replays. +- Aethercode accepts `signing_keys=...` for authenticated operations. Without + keys ordinary interpretation works, but signing/receipt export rejects and + verification returns false. `LEDGER SIGN` uses HMAC-SHA256 with a ledger domain. + `LEDGER VERIFY {"message":"hello","mac":"<64 hex characters>"}` checks that + exact message. Demonstrations explicitly create ephemeral random keys. + +These are deliberate compatibility breaks for insecure defaults and tags. +Provision keys from a secret store, update all communicating peers, and issue +fresh envelopes; do not add an acceptance fallback for the old format. The +`TONGUE_KEYS` symbol remains importable only as historical public demo data and +must not be used as a secret. No existing secret store is modified by this patch. + +`tests/security/test_rwp2_authentication_contract.py` covers both failures and +ordinary operation, all four tiers, Unicode, binary payloads, wrong keys, +tampered headers, malformed tags and concurrent replay. Together with the +canonical-registry and full-system checks, the focused follow-up had 39 passing +tests. The full-system test's broad title is not a certification claim. + ## Reproduction and validation `tests/security/test_braid_signature_integrity.py` exercises both shipped Python diff --git a/scripts/system/run_core_python_checks.py b/scripts/system/run_core_python_checks.py index c76c81a2d..501bb852a 100644 --- a/scripts/system/run_core_python_checks.py +++ b/scripts/system/run_core_python_checks.py @@ -31,6 +31,7 @@ "tests/governance/test_full_system_decision_precedence.py", "tests/security/test_breathing_and_backend_contracts.py", "tests/security/test_braid_signature_integrity.py", + "tests/security/test_rwp2_authentication_contract.py", "tests/test_sacred_eggs.py", "tests/test_sacred_egg_registry.py", "tests/test_semantic_projector_deep.py", diff --git a/src/crypto/sacred_eggs.py b/src/crypto/sacred_eggs.py index 8713e11cf..d01c66301 100644 --- a/src/crypto/sacred_eggs.py +++ b/src/crypto/sacred_eggs.py @@ -690,6 +690,10 @@ def create_session_egg( Derives the yolk deterministically from the session ID so the same session always produces the same egg. + This is an identifier helper, not secret credential generation. Anyone + knowing the session ID can derive the yolk; use an independently secret + random yolk for authentication. Hashing a public ID does not hide it. + Args: session_id: Unique session identifier context: Purpose context diff --git a/src/spiralverse/aethercode.py b/src/spiralverse/aethercode.py index 08cb881b7..b7c72c612 100644 --- a/src/spiralverse/aethercode.py +++ b/src/spiralverse/aethercode.py @@ -31,6 +31,9 @@ import hashlib import re +import hmac +import json +import secrets import struct import numpy as np from dataclasses import dataclass, field @@ -45,7 +48,7 @@ RWP2Envelope, EnvelopeFactory, OperationTier, - TONGUE_KEYS, + SignatureEngine, ) # ============================================================================= @@ -576,8 +579,18 @@ class LedgerHandler(DomainHandler): - PROOF : Generate proof """ - def __init__(self): + def __init__(self, signing_key: Optional[bytes] = None): super().__init__(TongueID.LEDGER) + if signing_key is not None: + SignatureEngine({ProtocolTongue.DR: signing_key}) + self._signing_key = signing_key + + def _mac(self, message: str) -> str: + if self._signing_key is None: + raise RuntimeError("A configured ledger authentication key is required") + return hmac.new( + self._signing_key, b"scbe:aether-ledger:v1\0" + message.encode("utf-8"), hashlib.sha256 + ).hexdigest() def execute(self, verse: AetherVerse, ctx: AetherContext) -> Any: content = verse.content.strip() @@ -598,11 +611,8 @@ def execute(self, verse: AetherVerse, ctx: AetherContext) -> Any: # SIGN command if content.startswith("SIGN"): message = content[4:].strip().strip("\"'") - # Simple HMAC-like signature - key = TONGUE_KEYS.get(ProtocolTongue.DR, b"default") - sig_input = f"{message}:{key.hex()}" - signature = hashlib.sha256(sig_input.encode()).hexdigest()[:32] - ctx.emit(f"[LEDGER] Signed: {signature[:16]}...") + signature = self._mac(message) + ctx.emit(f"[LEDGER] MAC generated: {signature[:16]}...") return signature # PROOF command @@ -621,10 +631,23 @@ def execute(self, verse: AetherVerse, ctx: AetherContext) -> Any: # VERIFY command if content.startswith("VERIFY"): - sig = content[6:].strip() - # Simplified verification (always true for demo) - ctx.emit(f"[LEDGER] Verified: {sig[:16]}...") - return True + # VERIFY {"message":"...","mac":"..."}; a tag alone is + # insufficient. This is shared-key authentication, not a signature. + verified = False + try: + request = json.loads(content[6:].strip()) + if ( + isinstance(request, dict) + and isinstance(request.get("message"), str) + and isinstance(request.get("mac"), str) + and request["mac"].isascii() + and len(request["mac"]) == 64 + ): + verified = hmac.compare_digest(request["mac"], self._mac(request["message"])) + except (ValueError, TypeError, RuntimeError): + verified = False + ctx.emit(f"[LEDGER] Verification: {'valid' if verified else 'invalid'}") + return verified return None @@ -767,14 +790,15 @@ class AethercodeInterpreter: - RWP2-signed execution proofs """ - def __init__(self, synthesize_audio: bool = True): + def __init__(self, synthesize_audio: bool = True, signing_keys: Optional[Dict[ProtocolTongue, bytes]] = None): + self._envelope_factory = EnvelopeFactory(keys=signing_keys) if signing_keys is not None else None self.handlers: Dict[TongueID, DomainHandler] = { TongueID.AXIOM: AxiomHandler(), TongueID.FLOW: FlowHandler(), TongueID.GLYPH: GlyphHandler(), TongueID.ORACLE: OracleHandler(), TongueID.CHARM: CharmHandler(), - TongueID.LEDGER: LedgerHandler(), + TongueID.LEDGER: LedgerHandler(signing_key=signing_keys.get(ProtocolTongue.DR) if signing_keys else None), } self.synthesize_audio = synthesize_audio self.synthesizer = ChantSynthesizer() if synthesize_audio else None @@ -831,8 +855,12 @@ def execute_source(self, source: str, title: str = "Untitled") -> AetherContext: def export_proof(self, ctx: AetherContext, filename: str = None) -> RWP2Envelope: """ - Generate RWP2-signed proof of execution. + Generate a shared-key authenticated receipt of the supplied trace. + + Authentication does not prove that the reported computation occurred. """ + if self._envelope_factory is None: + raise RuntimeError("Explicit signing_keys are required to authenticate execution receipts") # Build proof payload proof_data = { "verses": len(ctx.trace), @@ -844,7 +872,7 @@ def export_proof(self, ctx: AetherContext, filename: str = None) -> RWP2Envelope payload = str(proof_data).encode() # Create envelope with all tongues used - factory = EnvelopeFactory() + factory = self._envelope_factory tongues_used = [ (ProtocolTongue[t.value[:2].upper()] if hasattr(ProtocolTongue, t.value[:2].upper()) else ProtocolTongue.KO) for t in set(tv.tongue_id for tv in ctx.trace if hasattr(tv, "tongue_id")) @@ -953,7 +981,8 @@ def demo(): print() # Initialize interpreter - interpreter = AethercodeInterpreter(synthesize_audio=True) + demo_keys = {t: secrets.token_bytes(32) for t in ProtocolTongue} + interpreter = AethercodeInterpreter(synthesize_audio=True, signing_keys=demo_keys) # Demo 1: Hello World print("[DEMO 1] Hello World") @@ -966,7 +995,7 @@ def demo(): # Demo 2: Fibonacci print("[DEMO 2] Fibonacci Structure") print("-" * 60) - interpreter2 = AethercodeInterpreter(synthesize_audio=True) + interpreter2 = AethercodeInterpreter(synthesize_audio=True, signing_keys=demo_keys) ctx2 = interpreter2.execute_source(FIBONACCI, "Fibonacci") for line in ctx2.output: print(f" {line}") @@ -975,7 +1004,7 @@ def demo(): # Demo 3: Full Six-Tongue Demo print("[DEMO 3] Full Six-Tongue Composition") print("-" * 60) - interpreter3 = AethercodeInterpreter(synthesize_audio=True) + interpreter3 = AethercodeInterpreter(synthesize_audio=True, signing_keys=demo_keys) ctx3 = interpreter3.execute_source(FULL_DEMO, "Six Tongues Symphony") for line in ctx3.output: print(f" {line}") diff --git a/src/spiralverse/rwp2_envelope.py b/src/spiralverse/rwp2_envelope.py index 0e552a764..b3c2fbc8b 100644 --- a/src/spiralverse/rwp2_envelope.py +++ b/src/spiralverse/rwp2_envelope.py @@ -32,6 +32,7 @@ import json import secrets import time +import threading from dataclasses import dataclass, field from datetime import datetime from typing import Dict, List, Optional, Set, Tuple, Any @@ -64,8 +65,8 @@ class ProtocolTongue(Enum): DR = "DR" # Draumric - Types & Schema -# Tongue-specific HMAC keys (in production, these would be securely managed) -# Using deterministic seeds for reproducibility in this implementation +# Historical public demonstration values, retained only for compatibility of +# imports. SignatureEngine rejects these as authentication keys. TONGUE_KEYS = { ProtocolTongue.KO: hashlib.sha256(b"SCBE_KO_KEY_v1").digest(), ProtocolTongue.AV: hashlib.sha256(b"SCBE_AV_KEY_v1").digest(), @@ -85,8 +86,8 @@ class OperationTier(Enum): """ Tiered operation classification by risk level. - Security scales exponentially: S(N) = B * R^(N^2) - where B = base bits (256), R = harmonic ratio (1.5), N = tongue count + Tiers select mandatory signer roles. The geometric display multipliers + below are not estimates of cryptographic security bits. """ TIER_1 = 1 # Single tongue (KO) - Basic coordination @@ -325,22 +326,52 @@ class SignatureEngine: """ def __init__(self, keys: Optional[Dict[ProtocolTongue, bytes]] = None): - self.keys = keys or TONGUE_KEYS + if not isinstance(keys, dict) or not keys: + raise ValueError("Explicit nonempty tongue keys are required") + for tongue, key in keys.items(): + if not isinstance(tongue, ProtocolTongue) or not isinstance(key, bytes) or len(key) < 32: + raise ValueError("Tongue keys must be bytes of at least 32 bytes") + if key in TONGUE_KEYS.values(): + raise ValueError("Public demo keys cannot authenticate envelopes") + self.keys = dict(keys) def _compute_signature_input(self, envelope: RWP2Envelope) -> bytes: """ Compute the canonical input for signature. - Includes: spelltext + payload + aad + nonce + timestamp + Canonical, typed fields including routing/security metadata. Revision 2 + MACs deliberately do not verify the old delimiter-joined encoding. """ - parts = [ - envelope.spelltext.encode("utf-8"), - envelope.payload, - envelope.aad.encode("utf-8"), - envelope.nonce.encode("utf-8"), - str(envelope.timestamp_ms).encode("utf-8"), - ] - return b"|".join(parts) + if ( + envelope.version != "2" + or not isinstance(envelope.tier, OperationTier) + or not isinstance(envelope.payload, bytes) + or type(envelope.timestamp_ms) is not int + or envelope.timestamp_ms <= 0 + or not isinstance(envelope.signatures, dict) + or any(not isinstance(v, str) for v in (envelope.spelltext, envelope.aad, envelope.kid, envelope.nonce)) + or not envelope.nonce + ): + raise ValueError("Malformed RWP2 envelope") + fields = { + "domain": "scbe:rwp2:mac:v2", + "version": envelope.version, + "spelltext": envelope.spelltext, + "payload_b64": envelope.payload_b64, + "aad": envelope.aad, + "nonce": envelope.nonce, + "timestamp_ms": envelope.timestamp_ms, + "kid": envelope.kid, + "tier": envelope.tier.value, + } + return json.dumps(fields, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") + + def _mac(self, tongue: ProtocolTongue, sig_input: bytes) -> str: + return hmac.new( + self.keys[tongue], + b"scbe:rwp2:tongue:v2\0" + tongue.value.encode("ascii") + b"\0" + sig_input, + hashlib.sha256, + ).hexdigest() def sign(self, envelope: RWP2Envelope, tongues: Set[ProtocolTongue]) -> RWP2Envelope: """ @@ -349,13 +380,12 @@ def sign(self, envelope: RWP2Envelope, tongues: Set[ProtocolTongue]) -> RWP2Enve Returns new envelope with signatures added. """ sig_input = self._compute_signature_input(envelope) + if not tongues or any(t not in self.keys for t in tongues): + raise ValueError("Every requested signer requires a configured key") new_signatures = dict(envelope.signatures) for tongue in tongues: - key = self.keys.get(tongue) - if key: - sig = hmac.new(key, sig_input, hashlib.sha256).hexdigest() - new_signatures[tongue] = sig + new_signatures[tongue] = self._mac(tongue, sig_input) # Return new envelope with signatures return RWP2Envelope( @@ -385,22 +415,30 @@ def verify( Returns: (overall_valid, per_tongue_results) """ - sig_input = self._compute_signature_input(envelope) + try: + sig_input = self._compute_signature_input(envelope) + except (ValueError, TypeError, AttributeError): + return False, {} # Determine required tongues + tier_required = TIER_REQUIRED_TONGUES[envelope.tier] if required_tongues is None: - required_tongues = TIER_REQUIRED_TONGUES.get(envelope.tier, {ProtocolTongue.KO}) + required_tongues = set(tier_required) + if not required_tongues or not tier_required.issubset(required_tongues): + return False, {} + if any(not isinstance(t, ProtocolTongue) for t in required_tongues): + return False, {} results = {} for tongue in required_tongues: key = self.keys.get(tongue) sig = envelope.signatures.get(tongue) - if not key or not sig: + if not key or not isinstance(sig, str) or len(sig) != 64 or not sig.isascii(): results[tongue] = False continue - expected_sig = hmac.new(key, sig_input, hashlib.sha256).hexdigest() + expected_sig = self._mac(tongue, sig_input) results[tongue] = hmac.compare_digest(sig, expected_sig) # Overall valid only if ALL required tongues verify @@ -422,9 +460,17 @@ class ReplayProtector: """ def __init__(self, max_age_seconds: int = 300, max_cache_size: int = 10000): # 5 minutes + if ( + type(max_age_seconds) is not int + or max_age_seconds < 0 + or type(max_cache_size) is not int + or max_cache_size < 1 + ): + raise ValueError("Replay bounds must be nonnegative age and positive capacity") self.max_age = max_age_seconds self.max_cache = max_cache_size self.used_nonces: Dict[str, int] = {} # nonce -> timestamp_ms + self._lock = threading.Lock() def is_valid(self, envelope: RWP2Envelope) -> Tuple[bool, str]: """ @@ -432,7 +478,14 @@ def is_valid(self, envelope: RWP2Envelope) -> Tuple[bool, str]: Returns (valid, reason). """ + # This object is process-local; authenticate before calling it. + with self._lock: + return self._check_and_record(envelope) + + def _check_and_record(self, envelope: RWP2Envelope) -> Tuple[bool, str]: now_ms = int(time.time() * 1000) + if type(envelope.timestamp_ms) is not int or not isinstance(envelope.nonce, str) or not envelope.nonce: + return False, "Malformed freshness fields" # Check timestamp freshness age_ms = now_ms - envelope.timestamp_ms @@ -447,6 +500,10 @@ def is_valid(self, envelope: RWP2Envelope) -> Tuple[bool, str]: if nonce_key in self.used_nonces: return False, "Nonce already used (replay detected)" + # Never evict a still-valid receipt to admit a new one. + self._cleanup() + if len(self.used_nonces) >= self.max_cache: + return False, "Replay cache full" # Mark nonce as used self.used_nonces[nonce_key] = envelope.timestamp_ms @@ -465,14 +522,6 @@ def _cleanup(self): for k in to_remove: del self.used_nonces[k] - # Enforce cache size limit - if len(self.used_nonces) > self.max_cache: - # Remove oldest entries - sorted_items = sorted(self.used_nonces.items(), key=lambda x: x[1]) - remove_count = len(self.used_nonces) - self.max_cache - for k, _ in sorted_items[:remove_count]: - del self.used_nonces[k] - # ============================================================================= # Envelope Factory @@ -484,8 +533,8 @@ class EnvelopeFactory: Factory for creating and validating RWP2 envelopes. """ - def __init__(self): - self.signature_engine = SignatureEngine() + def __init__(self, keys: Optional[Dict[ProtocolTongue, bytes]] = None): + self.signature_engine = SignatureEngine(keys) self.replay_protector = ReplayProtector() self.sequence_counter = 0 @@ -534,17 +583,18 @@ def validate(self, envelope: RWP2Envelope) -> Tuple[bool, List[str]]: """ issues = [] + # Verify before consuming freshness state; unauthenticated traffic must + # not burn a legitimate message's nonce. + sig_valid, sig_results = self.signature_engine.verify(envelope) + if not sig_valid: + failed_tongues = [t.value for t, v in sig_results.items() if not v] + return False, [f"Signature verification failed for: {failed_tongues}"] + # Check replay protection replay_valid, replay_reason = self.replay_protector.is_valid(envelope) if not replay_valid: issues.append(f"Replay check failed: {replay_reason}") - # Verify signatures - sig_valid, sig_results = self.signature_engine.verify(envelope) - if not sig_valid: - failed_tongues = [t.value for t, v in sig_results.items() if not v] - issues.append(f"Signature verification failed for: {failed_tongues}") - return len(issues) == 0, issues @@ -560,7 +610,7 @@ def demo(): print("=" * 70) print() - factory = EnvelopeFactory() + factory = EnvelopeFactory(keys={t: secrets.token_bytes(32) for t in ProtocolTongue}) # Create a Tier 1 envelope (single tongue) print("[TIER 1] Basic coordination message (KO only):") diff --git a/tests/security/test_rwp2_authentication_contract.py b/tests/security/test_rwp2_authentication_contract.py new file mode 100644 index 000000000..06fff6ba0 --- /dev/null +++ b/tests/security/test_rwp2_authentication_contract.py @@ -0,0 +1,165 @@ +"""Legacy shared-key envelope authentication and interpreter caller contracts.""" + +from dataclasses import replace +import hashlib +import hmac +import json +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from src.spiralverse.rwp2_envelope import ( + EnvelopeFactory, + OperationTier, + ProtocolTongue, + ReplayProtector, + RWP2Envelope, + SignatureEngine, + TONGUE_KEYS, +) +from src.spiralverse.aethercode import AethercodeInterpreter, AetherContext, AetherVerse, LedgerHandler +from src.spiralverse.polyglot_alphabet import TongueID + + +@pytest.fixture +def keys(): + # Public test fixtures; never production keys. + return {t: hashlib.sha256(b"rwp2-test-only:" + t.value.encode()).digest() for t in ProtocolTongue} + + +@pytest.mark.parametrize("keys", [None, {}, TONGUE_KEYS]) +def test_missing_or_public_demo_keys_reject(keys): + with pytest.raises(ValueError): + SignatureEngine(keys) + + +@pytest.mark.parametrize("field,value", [("kid", "other"), ("version", "3"), ("tier", OperationTier.TIER_2)]) +def test_security_metadata_is_bound(keys, field, value): + engine = SignatureEngine(keys) + signed = engine.sign(RWP2Envelope(payload=b"test"), set(ProtocolTongue)) + assert engine.verify(signed)[0] + assert not engine.verify(replace(signed, **{field: value}))[0] + + +def test_delimiter_shift_does_not_preserve_authentication(keys): + engine = SignatureEngine(keys) + signed = engine.sign(RWP2Envelope(spelltext="a|b", payload=b"c"), {ProtocolTongue.KO}) + altered = replace(signed, spelltext="a", payload=b"b|c") + assert not engine.verify(altered)[0] + + +def test_explicit_tongues_cannot_weaken_tier_or_be_empty(keys): + engine = SignatureEngine(keys) + signed = engine.sign(RWP2Envelope(tier=OperationTier.TIER_3), {ProtocolTongue.KO}) + assert not engine.verify(signed, set())[0] + assert not engine.verify(signed, {ProtocolTongue.KO})[0] + + +def test_distinct_tongues_are_domain_separated(keys): + # Even an accidentally reused key must not allow retagging as another tongue. + shared = {t: keys[ProtocolTongue.KO] for t in ProtocolTongue} + engine = SignatureEngine(shared) + signed = engine.sign(RWP2Envelope(tier=OperationTier.TIER_2), {ProtocolTongue.KO}) + signed.signatures[ProtocolTongue.RU] = signed.signatures[ProtocolTongue.KO] + assert not engine.verify(signed)[0] + + +def test_signing_without_requested_key_rejects(keys): + engine = SignatureEngine({ProtocolTongue.KO: keys[ProtocolTongue.KO]}) + with pytest.raises(ValueError): + engine.sign(RWP2Envelope(), {ProtocolTongue.RU}) + + +def test_keys_are_copied_and_roundtrip_works(keys): + engine = SignatureEngine(keys) + signed = engine.sign(RWP2Envelope(payload=b"\x00\xff|", aad="a|b"), {ProtocolTongue.KO}) + keys[ProtocolTongue.KO] = b"x" * 32 + assert engine.verify(RWP2Envelope.from_json(signed.to_json()))[0] + + +def test_bad_mac_cannot_consume_a_valid_messages_nonce(keys): + factory = EnvelopeFactory(keys=keys) + signed = factory.create("TEST", b"test", ProtocolTongue.KO, OperationTier.TIER_1) + assert not factory.validate(replace(signed, payload=b"tampered"))[0] + assert factory.validate(signed)[0] + assert not factory.validate(signed)[0] + + +def test_replay_cache_cannot_evict_live_receipts(): + replay = ReplayProtector(max_cache_size=1) + first, second = RWP2Envelope(), RWP2Envelope() + assert replay.is_valid(first)[0] + assert not replay.is_valid(second)[0] + assert not replay.is_valid(first)[0] + + +def test_ledger_never_verifies_without_message_and_key(): + handler = LedgerHandler() + assert handler.execute(AetherVerse(TongueID.LEDGER, "test", "VERIFY nonsense"), AetherContext()) is False + + +def test_ledger_mac_roundtrip_and_wrong_message(keys): + key = keys[ProtocolTongue.DR] + handler = LedgerHandler(signing_key=key) + ctx = AetherContext() + signature = handler.execute(AetherVerse(TongueID.LEDGER, "test", 'SIGN "hello"'), ctx) + assert signature == hmac.new(key, b"scbe:aether-ledger:v1\0hello", hashlib.sha256).hexdigest() + request = json.dumps({"message": "hello", "mac": signature}) + assert handler.execute(AetherVerse(TongueID.LEDGER, "test", "VERIFY " + request), ctx) is True + request = json.dumps({"message": "changed", "mac": signature}) + assert handler.execute(AetherVerse(TongueID.LEDGER, "test", "VERIFY " + request), ctx) is False + + +def test_interpreter_proof_requires_keys_and_verifies_with_matching_keys(keys): + with pytest.raises(RuntimeError): + AethercodeInterpreter(synthesize_audio=False).export_proof(AetherContext()) + interpreter = AethercodeInterpreter(synthesize_audio=False, signing_keys=keys) + signed = interpreter.export_proof(AetherContext()) + assert SignatureEngine(keys).verify(signed)[0] + + +@pytest.mark.parametrize("tier", list(OperationTier)) +def test_all_tiers_with_unicode_delimiters_and_binary_payload(keys, tier): + factory = EnvelopeFactory(keys) + signed = factory.create("TEST", b"\x00|\xff", ProtocolTongue.KO, tier, aad="unicode: \u03c6|\u96ea") + restored = RWP2Envelope.from_json(signed.to_json()) + assert factory.validate(restored)[0] + + +@pytest.mark.parametrize( + "field,value", + [("nonce", "other"), ("timestamp_ms", 1), ("aad", "other"), ("spelltext", "other"), ("payload", b"other")], +) +def test_every_content_field_is_bound(keys, field, value): + engine = SignatureEngine(keys) + signed = engine.sign(RWP2Envelope(payload=b"original"), {ProtocolTongue.KO}) + assert not engine.verify(replace(signed, **{field: value}))[0] + + +@pytest.mark.parametrize("bad_tag", [None, b"x" * 64, "\u00e9" * 64, "x" * 64, ""]) +def test_malformed_tags_return_false(keys, bad_tag): + envelope = RWP2Envelope(signatures={ProtocolTongue.KO: bad_tag}) + assert not SignatureEngine(keys).verify(envelope)[0] + + +def test_concurrent_replay_has_one_winner(keys): + factory = EnvelopeFactory(keys) + signed = factory.create("TEST", b"test", ProtocolTongue.KO, OperationTier.TIER_1) + with ThreadPoolExecutor(max_workers=4) as executor: + verdicts = list(executor.map(lambda _: factory.validate(signed)[0], range(12))) + assert sum(verdicts) == 1 + + +def test_non_signing_interpreter_still_runs(): + context = AethercodeInterpreter(synthesize_audio=False).execute_source('a3f7c2e1:EMIT "Hello"') + assert any("Hello" in line for line in context.output) + assert not any("[ERROR]" in line for line in context.output) + + +def test_ledger_wrong_key_rejects(keys): + signer = LedgerHandler(signing_key=keys[ProtocolTongue.DR]) + verifier = LedgerHandler(signing_key=keys[ProtocolTongue.KO]) + ctx = AetherContext() + tag = signer.execute(AetherVerse(TongueID.LEDGER, "test", 'SIGN "hello"'), ctx) + request = json.dumps({"message": "hello", "mac": tag}) + assert verifier.execute(AetherVerse(TongueID.LEDGER, "test", "VERIFY " + request), ctx) is False diff --git a/tests/test_full_system_verification.py b/tests/test_full_system_verification.py index 38da46f2e..eebe00162 100644 --- a/tests/test_full_system_verification.py +++ b/tests/test_full_system_verification.py @@ -467,7 +467,9 @@ def decide(risk): check("RWP2 envelope imports", True) - sig_engine = SignatureEngine() + import secrets + + sig_engine = SignatureEngine({t: secrets.token_bytes(32) for t in PT}) # Create and sign envelope env = RWP2Envelope( From aa60e29177e35055af2ebee58feb3cab6f73ecf7 Mon Sep 17 00:00:00 2001 From: issdandavis <215328633+issdandavis@users.noreply.github.com> Date: Fri, 18 Sep 2026 17:59:48 -0700 Subject: [PATCH 3/3] ci(crypto): install signing adapter dependency in native smoke --- .github/workflows/pqc-native-liboqs.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/pqc-native-liboqs.yml b/.github/workflows/pqc-native-liboqs.yml index 26a460089..5d5c0e77a 100644 --- a/.github/workflows/pqc-native-liboqs.yml +++ b/.github/workflows/pqc-native-liboqs.yml @@ -88,7 +88,7 @@ jobs: - name: Install Python native bindings run: | python -m pip install --upgrade pip - python -m pip install liboqs-python==0.16.0 + python -m pip install liboqs-python==0.16.0 cryptography==50.0.0 - name: Check native backend and reject tampered signatures (not FIPS validation) run: python scripts/security/native_liboqs_smoke.py