[P0][A3] Replace base64 stub with real AES-256-GCM encryption in Vault

[POWDER-RANGER]

Problem

vault/vault.py uses base64 encoding as a fake encryption layer. The ENHANCEMENT_ROADMAP.md explicitly acknowledges this: "uses base64 (NOT encryption!)". This is a critical security failure — the entire framework's security proposition rests on vault secrecy, but any stored value is trivially recoverable with base64.b64decode().

Current behavior: _encrypt(value) returns base64.b64encode(value.encode()).decode() — fully reversible, zero secrecy, zero integrity.

Required behavior: AES-256-GCM authenticated encryption where:

• Stored values are not recoverable without the 256-bit key
• Tampered ciphertext causes InvalidTag exception (authentication)
• Nonce is random and unique per encryption call
• Tag is verified before any plaintext is returned

Implementation

Core Crypto (using cryptography library — already in requirements.txt)

# oblisk/vault/crypto.py
import os
import struct
from base64 import b64decode, b64encode
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes
from cryptography.exceptions import InvalidTag

NONCE_SIZE = 12  # 96-bit nonce for AES-GCM (NIST recommendation)
KEY_SIZE = 32    # 256-bit key
SALT_SIZE = 16   # 128-bit salt for PBKDF2
PBKDF2_ITERATIONS = 600_000  # OWASP 2023 recommendation


def derive_key(passphrase: str, salt: bytes | None = None) -> tuple[bytes, bytes]:
    """Derive a 256-bit key from a passphrase using PBKDF2-HMAC-SHA256.
    Returns (key, salt). Salt is generated if not provided.
    """
    if salt is None:
        salt = os.urandom(SALT_SIZE)
    kdf = PBKDF2HMAC(
        algorithm=hashes.SHA256(),
        length=KEY_SIZE,
        salt=salt,
        iterations=PBKDF2_ITERATIONS,
    )
    key = kdf.derive(passphrase.encode())
    return key, salt


def encrypt(key: bytes, plaintext: str) -> bytes:
    """Encrypt plaintext using AES-256-GCM.
    Returns: nonce (12 bytes) || ciphertext+tag (variable)
    """
    if len(key) != KEY_SIZE:
        raise ValueError(f"Key must be exactly {KEY_SIZE} bytes, got {len(key)}")
    nonce = os.urandom(NONCE_SIZE)
    aesgcm = AESGCM(key)
    ciphertext_and_tag = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
    return nonce + ciphertext_and_tag


def decrypt(key: bytes, blob: bytes) -> str:
    """Decrypt AES-256-GCM blob. Raises InvalidTag if tampered."""
    if len(key) != KEY_SIZE:
        raise ValueError(f"Key must be exactly {KEY_SIZE} bytes, got {len(key)}")
    if len(blob) < NONCE_SIZE:
        raise ValueError("Ciphertext blob too short")
    nonce = blob[:NONCE_SIZE]
    ciphertext_and_tag = blob[NONCE_SIZE:]
    aesgcm = AESGCM(key)
    try:
        plaintext_bytes = aesgcm.decrypt(nonce, ciphertext_and_tag, None)
    except InvalidTag:
        raise InvalidTag("Vault decryption failed: data may be tampered or key is wrong")
    return plaintext_bytes.decode("utf-8")


def encode_blob(blob: bytes) -> str:
    """Base64-encode raw ciphertext blob for storage (JSON-safe)."""
    return b64encode(blob).decode("ascii")


def decode_blob(encoded: str) -> bytes:
    """Decode stored ciphertext blob."""
    return b64decode(encoded)

Vault Class Rewrite

# oblisk/vault/vault.py
import json
import os
from pathlib import Path
from typing import Any
from cryptography.exceptions import InvalidTag
from oblisk.vault.crypto import derive_key, encrypt, decrypt, encode_blob, decode_blob, KEY_SIZE


class Vault:
    """AES-256-GCM encrypted key-value store.
    
    Args:
        key: 32-byte symmetric key. Use derive_key() to generate from passphrase.
        path: Optional file path for persistent storage.
    """
    
    def __init__(self, key: bytes, path: str | Path | None = None) -> None:
        if len(key) != KEY_SIZE:
            raise ValueError(f"Vault key must be {KEY_SIZE} bytes (256-bit). Got {len(key)}.")
        self._key = key
        self._path = Path(path) if path else None
        self._store: dict[str, str] = {}  # stores encoded (encrypted) blobs
        
        if self._path and self._path.exists():
            self._load()
    
    def store(self, name: str, value: str) -> None:
        """Encrypt and store a value."""
        if not isinstance(value, str):
            raise TypeError(f"Vault only stores str values, got {type(value).__name__}")
        blob = encrypt(self._key, value)
        self._store[name] = encode_blob(blob)
        if self._path:
            self._save()
    
    def retrieve(self, name: str) -> str:
        """Retrieve and decrypt a value. Raises KeyError if not found, InvalidTag if tampered."""
        if name not in self._store:
            raise KeyError(f"Secret '{name}' not found in vault")
        blob = decode_blob(self._store[name])
        return decrypt(self._key, blob)
    
    def delete(self, name: str) -> None:
        """Remove a secret from the vault."""
        if name not in self._store:
            raise KeyError(f"Secret '{name}' not found in vault")
        del self._store[name]
        if self._path:
            self._save()
    
    def list_secrets(self) -> list[str]:
        """Return list of stored secret names (never values)."""
        return list(self._store.keys())
    
    def __contains__(self, name: str) -> bool:
        return name in self._store
    
    def __len__(self) -> int:
        return len(self._store)
    
    def _save(self) -> None:
        """Persist encrypted store to disk."""
        self._path.parent.mkdir(parents=True, exist_ok=True)
        self._path.write_text(json.dumps(self._store, indent=2))
    
    def _load(self) -> None:
        """Load encrypted store from disk."""
        data = json.loads(self._path.read_text())
        if not isinstance(data, dict):
            raise ValueError("Vault file is corrupted")
        self._store = data

Tasks

• Create oblisk/vault/crypto.py with AES-256-GCM functions (as above)
• Rewrite oblisk/vault/vault.py to use real encryption (as above)
• Add derive_key() helper exported from oblisk.vault
• Update oblisk/vault/__init__.py to export Vault, derive_key, InvalidTag
• Add Vault.rotate_key(new_key: bytes) method (re-encrypt all values under new key)
• Add Vault.export_encrypted(path: Path) and Vault.import_encrypted(path: Path, key: bytes) for backup
• Verify cryptography is pinned in pyproject.toml with >=42.0 (supports AES-GCM natively)
• Write tests/unit/test_vault_crypto.py (see test spec below)
• Update examples/vault_demo.py to show key derivation + store/retrieve workflow
• Update docs/ARCHITECTURE.md to mark AES-256-GCM as Implemented (not Planned)

Test Spec

# tests/unit/test_vault_crypto.py
import pytest
from cryptography.exceptions import InvalidTag
from oblisk.vault.crypto import encrypt, decrypt, derive_key, KEY_SIZE
from oblisk.vault import Vault

class TestCryptoLayer:
    def test_encrypt_decrypt_roundtrip(self):
        key = b'a' * KEY_SIZE
        plaintext = "super-secret-value"
        blob = encrypt(key, plaintext)
        assert decrypt(key, blob) == plaintext

    def test_nonces_are_unique(self):
        key = b'a' * KEY_SIZE
        blobs = [encrypt(key, "same") for _ in range(100)]
        nonces = [b[:12] for b in blobs]
        assert len(set(nonces)) == 100  # all unique

    def test_tamper_raises_invalid_tag(self):
        key = b'a' * KEY_SIZE
        blob = bytearray(encrypt(key, "secret"))
        blob[15] ^= 0xFF  # flip a byte in ciphertext
        with pytest.raises(InvalidTag):
            decrypt(key, bytes(blob))

    def test_wrong_key_raises_invalid_tag(self):
        key1 = b'a' * KEY_SIZE
        key2 = b'b' * KEY_SIZE
        blob = encrypt(key1, "secret")
        with pytest.raises(InvalidTag):
            decrypt(key2, blob)

    def test_key_length_enforced(self):
        with pytest.raises(ValueError, match="32 bytes"):
            encrypt(b'short', "value")

    def test_derive_key_deterministic_with_salt(self):
        key1, salt = derive_key("passphrase")
        key2, _ = derive_key("passphrase", salt)
        assert key1 == key2

    def test_derive_key_different_salts(self):
        key1, _ = derive_key("passphrase")
        key2, _ = derive_key("passphrase")
        assert key1 != key2  # different random salts

class TestVault:
    @pytest.fixture
    def vault(self):
        return Vault(key=b'a' * KEY_SIZE)

    def test_store_retrieve(self, vault):
        vault.store("api_key", "sk-abc123")
        assert vault.retrieve("api_key") == "sk-abc123"

    def test_missing_key_raises_key_error(self, vault):
        with pytest.raises(KeyError):
            vault.retrieve("nonexistent")

    def test_delete(self, vault):
        vault.store("temp", "value")
        vault.delete("temp")
        assert "temp" not in vault

    def test_list_secrets(self, vault):
        vault.store("a", "1")
        vault.store("b", "2")
        secrets = vault.list_secrets()
        assert "a" in secrets
        assert "b" in secrets
        assert "1" not in secrets  # values never leaked

    def test_stored_values_not_plaintext(self, vault):
        vault.store("password", "hunter2")
        # Internal store should never contain plaintext
        assert "hunter2" not in str(vault._store)

    def test_invalid_key_length(self):
        with pytest.raises(ValueError, match="32 bytes"):
            Vault(key=b'tooshort')

    def test_persistence(self, tmp_path):
        key = b'a' * KEY_SIZE
        path = tmp_path / "vault.json"
        v1 = Vault(key=key, path=path)
        v1.store("persist", "survives")
        v2 = Vault(key=key, path=path)  # reload from disk
        assert v2.retrieve("persist") == "survives"

    def test_persistence_wrong_key(self, tmp_path):
        path = tmp_path / "vault.json"
        v1 = Vault(key=b'a' * KEY_SIZE, path=path)
        v1.store("secret", "value")
        v2 = Vault(key=b'b' * KEY_SIZE, path=path)  # wrong key
        with pytest.raises(InvalidTag):
            v2.retrieve("secret")

    def test_rotate_key(self, vault):
        vault.store("key", "value")
        new_key = b'b' * KEY_SIZE
        vault.rotate_key(new_key)
        assert vault.retrieve("key") == "value"  # still accessible

Acceptance Criteria

• ✅ vault.store("x", "secret") then "secret" not in str(vault._store) (plaintext never stored)
• ✅ vault.retrieve("x") returns original string after roundtrip
• ✅ Mutating stored ciphertext by 1 byte causes InvalidTag on retrieve
• ✅ Using a different key causes InvalidTag on retrieve
• ✅ 100 consecutive encryptions of the same value produce 100 unique blobs (nonce uniqueness)
• ✅ Key length validation enforced at Vault.__init__ and encrypt()
• ✅ Vault persists and reloads correctly across instances with the same key
• ✅ All 14 tests in test spec above pass
• ✅ docs/ARCHITECTURE.md marks vault as Implemented not Planned

Security Notes

This aligns with your OBELISK_CORE design doc:

• AES-256-GCM provides both confidentiality (AES) and integrity (GCM tag)
• PBKDF2-HMAC-SHA256 at 600k iterations matches OWASP 2023 recommendation
• 96-bit random nonces per NIST SP 800-38D
• cryptography library uses OpenSSL FIPS-validated implementation

Priority: P0 - Core security claim, critical credibility fix
Labels: security, vault, crypto, P0