open-SECURITY-{SOC_TYPE_II}-integration-fix

aether_protocol_c/audit.py

@@ -43,6 +43,7 @@
 import hashlib
 import json
 import sqlite3
+import threading
 import time
 from dataclasses import dataclass
 from pathlib import Path
@@ -93,15 +94,40 @@ def to_json(self) -> dict:
 
     @staticmethod
     def from_dict(d: dict) -> "AuditEntry":
-        """Reconstruct from dict."""
-        return AuditEntry(
-            timestamp=d["timestamp"],
-            phase=d["phase"],
-            order_id=d["order_id"],
-            data=d["data"],
-            signature=d["signature"],
-            quantum_proof=d["quantum_proof"],
-        )
+        """Reconstruct from dict.
+
+        Raises:
+            AuditError: if required keys are missing or the `data`,
+                `signature`, or `quantum_proof` fields are not dicts.
+                This keeps malformed/tampered JSONL lines from silently
+                propagating non-dict values into downstream verification
+                code, which would otherwise crash with an unhandled
+                AttributeError instead of a clean tamper report.
+        """
+        try:
+            data = d["data"]
+            signature = d["signature"]
+            quantum_proof = d["quantum_proof"]
+            entry = AuditEntry(
+                timestamp=d["timestamp"],
+                phase=d["phase"],
+                order_id=d["order_id"],
+                data=data,
+                signature=signature,
+                quantum_proof=quantum_proof,
+            )
+        except KeyError as exc:
+            raise AuditError(f"Malformed audit entry: missing key {exc}") from exc
+
+        for field_name in ("data", "signature", "quantum_proof"):
+            value = getattr(entry, field_name)
+            if not isinstance(value, dict):
+                raise AuditError(
+                    f"Malformed audit entry: field '{field_name}' must be a "
+                    f"dict, got {type(value).__name__}"
+                )
+
+        return entry
 
 
 def _extract_quantum_proof(data: dict) -> dict:
@@ -177,6 +203,12 @@ def __init__(
         # Set before _init_db so close()/__del__ are safe even if init fails.
         self._conn = None
 
+        # Guards the append/rotate critical section (JSONL write + SQLite
+        # index write + _line_count read-modify-write) so concurrent
+        # threads sharing one AuditLog (check_same_thread=False signals
+        # this is expected) cannot race on the same line number / offset.
+        self._append_lock = threading.RLock()
+
         # Initialise SQLite index
         self._init_db()
 
@@ -274,9 +306,12 @@ def _rebuild_index(self) -> None:
                     entry = AuditEntry.from_dict(data)
                     self._index_entry(entry, offset, line_num)
                     line_num += 1
-                except (json.JSONDecodeError, KeyError):
-                    line_num += 1
-                    continue
+                except (json.JSONDecodeError, KeyError) as exc:
+                    self._conn.commit()
+                    raise AuditError(
+                        f"Corrupt audit log entry at line {line_num} "
+                        f"while rebuilding index: {exc}"
+                    ) from exc
         self._conn.commit()
 
     def _index_entry(
@@ -387,24 +422,28 @@ def _append(self, entry: AuditEntry) -> None:
         Writes to the JSONL file (binary mode for reliable byte offsets)
         and indexes the entry in SQLite.
         """
-        # Check rotation before writing (but not for rotation entries
-        # themselves, to avoid infinite recursion)
-        if entry.phase != "LOG_ROTATION":
-            self._maybe_rotate()
-
-        line = json.dumps(
-            entry.to_json(), sort_keys=True, separators=(",", ":")
-        )
+        # Serialize the whole read-modify-write critical section: rotation
+        # check, JSONL append, SQLite index write, and _line_count bump
+        # must be atomic w.r.t. other threads sharing this AuditLog.
+        with self._append_lock:
+            # Check rotation before writing (but not for rotation entries
+            # themselves, to avoid infinite recursion)
+            if entry.phase != "LOG_ROTATION":
+                self._maybe_rotate()
+
+            line = json.dumps(
+                entry.to_json(), sort_keys=True, separators=(",", ":")
+            )
+
+            # Write to JSONL in binary mode for reliable byte offsets
+            with open(self._path, "ab") as f:
+                offset = f.tell()
+                f.write((line + "\n").encode("utf-8"))
 
-        # Write to JSONL in binary mode for reliable byte offsets
-        with open(self._path, "ab") as f:
-            offset = f.tell()
-            f.write((line + "\n").encode("utf-8"))
-
-        # Index in SQLite
-        self._index_entry(entry, offset, self._line_count)
-        self._conn.commit()
-        self._line_count += 1
+            # Index in SQLite
+            self._index_entry(entry, offset, self._line_count)
+            self._conn.commit()
+            self._line_count += 1
 
     def append_commitment(
         self, commitment: dict, signature: dict
@@ -526,8 +565,6 @@ def get_trade_flow(self, order_id: str) -> dict:
         Returns:
             Dict with data and signature for each phase.
         """
-        entries = self.read_by_order_id(order_id)
-
         flow: Dict[str, Any] = {
             "order_id": order_id,
             "commitment": None,
@@ -541,19 +578,44 @@ def get_trade_flow(self, order_id: str) -> dict:
             "settlement_quantum_proof": None,
         }
 
-        for entry in entries:
-            if entry.phase == PHASE_COMMITMENT:
-                flow["commitment"] = entry.data
-                flow["commitment_sig"] = entry.signature
-                flow["commitment_quantum_proof"] = entry.quantum_proof
-            elif entry.phase == PHASE_EXECUTION:
-                flow["execution"] = entry.data
-                flow["execution_sig"] = entry.signature
-                flow["execution_quantum_proof"] = entry.quantum_proof
-            elif entry.phase == PHASE_SETTLEMENT:
-                flow["settlement"] = entry.data
-                flow["settlement_sig"] = entry.signature
-                flow["settlement_quantum_proof"] = entry.quantum_proof
+        phase_to_keys = {
+            PHASE_COMMITMENT: ("commitment", "commitment_sig", "commitment_quantum_proof"),
+            PHASE_EXECUTION: ("execution", "execution_sig", "execution_quantum_proof"),
+            PHASE_SETTLEMENT: ("settlement", "settlement_sig", "settlement_quantum_proof"),
+        }
+        keys_by_record_id = {
+            f"{order_id}_{phase}": keys for phase, keys in phase_to_keys.items()
+        }
+
+        # Single indexed query + single file open for all three phases,
+        # instead of three separate get_by_id() round-trips -- still O(1)
+        # indexed lookups (no full-file scan), just batched.
+        placeholders = ",".join("?" * len(keys_by_record_id))
+        cur = self._conn.execute(
+            f"SELECT record_id, jsonl_offset FROM audit_index WHERE record_id IN ({placeholders})",
+            list(keys_by_record_id),
+        )
+        offsets_by_record_id = {row[0]: row[1] for row in cur.fetchall()}
+
+        if offsets_by_record_id:
+            with open(self._path, "rb") as f:
+                for record_id, offset in offsets_by_record_id.items():
+                    data_key, sig_key, proof_key = keys_by_record_id[record_id]
+                    f.seek(offset)
+                    raw = f.readline()
+                    if not raw:
+                        continue
+                    try:
+                        record = json.loads(raw.decode("utf-8"))
+                    except (json.JSONDecodeError, UnicodeDecodeError):
+                        continue
+                    try:
+                        entry = AuditEntry.from_dict(record)
+                    except (KeyError, TypeError) as exc:
+                        raise AuditError(f"Corrupt audit log entry: {exc}") from exc
+                    flow[data_key] = entry.data
+                    flow[sig_key] = entry.signature
+                    flow[proof_key] = entry.quantum_proof
 
         return flow
 

aether_protocol_c/cli.py

@@ -39,16 +39,44 @@
     "audit_log": "audit/audit.jsonl",
 }
 
+# Maximum number of bytes accepted for a single JSON input (file or stdin).
+# Guards against unbounded-memory DoS from a runaway pipe or malicious input.
+MAX_JSON_INPUT_BYTES = 10 * 1024 * 1024  # 10 MiB
+
 
 # ── IO helpers ────────────────────────────────────────────────────────────────
 
+def _read_limited(fh, max_bytes: int) -> str:
+    """Read at most `max_bytes` (+1 chunk) from a text file-like object.
+
+    Raises ValueError if the stream contains more than `max_bytes` of data,
+    instead of buffering an unbounded amount of input into memory.
+    """
+    chunks: list[str] = []
+    total = 0
+    # Read in bounded chunks rather than fh.read() so we never buffer more
+    # than max_bytes + one chunk's worth of attacker-controlled data.
+    chunk_size = 65536
+    while total <= max_bytes:
+        chunk = fh.read(chunk_size)
+        if not chunk:
+            break
+        chunks.append(chunk)
+        total += len(chunk)
+    if total > max_bytes:
+        raise ValueError(
+            f"JSON input exceeds maximum allowed size of {max_bytes} bytes"
+        )
+    return "".join(chunks)
+
+
 def _read_json(path: str | None) -> Any:
     """Read JSON from a file path, or from stdin when path is None or '-'."""
     if path in (None, "-"):
-        raw = sys.stdin.read()
+        raw = _read_limited(sys.stdin, MAX_JSON_INPUT_BYTES)
     else:
         with open(path, "r", encoding="utf-8") as fh:
-            raw = fh.read()
+            raw = _read_limited(fh, MAX_JSON_INPUT_BYTES)
     if not raw.strip():
         raise ValueError("no JSON input provided")
     return json.loads(raw)

aether_protocol_c/crypto.py

@@ -24,10 +24,13 @@
 
 import hashlib
 import json
+import logging
 import time
 from dataclasses import dataclass, field, asdict
 from typing import Any, Dict, Optional, Tuple
 
+logger = logging.getLogger(__name__)
+
 from .ephemeral_signer import EphemeralSigner
 
 
@@ -263,12 +266,10 @@ def verify(self, message: dict, signature: dict) -> bool:
         Returns:
             True if the signature is valid.
         """
-        # EphemeralSigner.verify uses the pubkey from the signature envelope,
-        # not the private key, so we can use a temporary signer for verification.
-        temp = EphemeralSigner(quantum_seed=1)  # seed irrelevant for verify
-        result = temp.verify(message, signature)
-        temp.destroy()
-        return result
+        # Delegates to the module-level verify_signature() (which itself
+        # delegates to EphemeralSigner.verify_static()) so there is a single
+        # implementation of signature verification in this package.
+        return verify_signature(message, signature)
 
 
 # ── Helper functions ──────────────────────────────────────────────────────────
@@ -303,11 +304,29 @@ def verify_signature(message: dict, signature: dict) -> bool:
         True if the signature is valid.
     """
     try:
-        temp = EphemeralSigner(quantum_seed=1)
-        result = temp.verify(message, signature)
-        temp.destroy()
-        return result
-    except Exception:
+        # verify_static() only parses the pubkey embedded in the signature
+        # envelope -- no private key is derived, unlike constructing a
+        # throwaway EphemeralSigner just to call its instance verify().
+        # This try/except is a deliberate second fail-closed layer, not
+        # pure duplication of verify_static's own -- it also catches
+        # unexpected failures at the call boundary itself (e.g. a caller
+        # substituting a broken verify_static implementation).
+        return EphemeralSigner.verify_static(message, signature)
+    except (KeyError, ValueError, TypeError) as exc:
+        # Malformed signature envelope (missing field, bad hex, wrong
+        # length, etc.) -- no key material is logged.
+        logger.debug(
+            "verify_signature() failed to parse signature envelope: %s: %s",
+            type(exc).__name__,
+            exc,
+        )
+        return False
+    except Exception as exc:
+        logger.debug(
+            "verify_signature() failed with unexpected error: %s: %s",
+            type(exc).__name__,
+            exc,
+        )
         return False