218 lines
9.5 KiB
Python
218 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Smoke-test client-wrapped project secret setup and rotation payload generation."""
|
|
|
|
import base64
|
|
import importlib.util
|
|
import pathlib
|
|
import sys
|
|
|
|
|
|
PROJECT_ID = "11111111-1111-1111-1111-111111111111"
|
|
ACCOUNT_ID = "22222222-2222-2222-2222-222222222222"
|
|
PASSPHRASE = "correct horse battery staple"
|
|
|
|
|
|
def load_module():
|
|
script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py"
|
|
sys.path.insert(0, str(script_path.parent))
|
|
spec = importlib.util.spec_from_file_location("napseer_mcp_server", script_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def assert_wrapped_record(mod, record, secret_kind, data_key_epoch=1):
|
|
assert record["schema"] == mod.WRAPPED_PROJECT_KEY_BUNDLE_SCHEMA
|
|
assert record["project_id"] == PROJECT_ID
|
|
assert record["account_id"] == ACCOUNT_ID
|
|
assert record["kdf"] == "PBKDF2-HMAC-SHA256"
|
|
assert record["kdf_params"]["iterations"] >= 600000
|
|
assert record["expansion"] == "HKDF-SHA256"
|
|
assert f":kind:{secret_kind}:" in record["hkdf_info"]
|
|
assert record["aead_alg"] == "AES-256-GCM"
|
|
assert record["aad_canonicalization"] == "json-sorted-keys-v1"
|
|
assert record["status"] == "active"
|
|
assert record["data_key_epoch"] == data_key_epoch
|
|
assert len(base64.b64decode(record["salt_b64"], validate=True)) >= 16
|
|
assert len(base64.b64decode(record["nonce_b64"], validate=True)) == 12
|
|
bundle = mod.unwrap_project_key_bundle(record)
|
|
assert set(bundle["keys"]) == {"chat", "tabs", "gateway", "memory"}
|
|
for kind, state in bundle["keys"].items():
|
|
version = str(state["active_version"])
|
|
key = base64.b64decode(state["versions"][version]["key_b64"], validate=True)
|
|
assert len(key) == 32, kind
|
|
|
|
|
|
def run():
|
|
mod = load_module()
|
|
writes = []
|
|
|
|
mod.AUTH = {"agent_id": "gateway-agent-1", "account_id": ACCOUNT_ID}
|
|
mod.GATEWAY_PROJECT_BUNDLE_PASSPHRASE = PASSPHRASE
|
|
mod.VAULT_KEY = bytes(range(32))
|
|
mod.VAULT_SECRETS = {}
|
|
mod.persist_vault_secrets = lambda: writes.append(dict(mod.VAULT_SECRETS))
|
|
|
|
payload = mod.gateway_generate_project_vault_setup_payload(PROJECT_ID, "setup-1")
|
|
assert mod.VAULT_KDF_ITERATIONS == 600000
|
|
assert mod.PROJECT_BUNDLE_KDF_ITERATIONS >= 600000
|
|
assert not writes, "project secret payload generation must not persist local vault state"
|
|
assert set(payload) == {"project_secrets"}
|
|
assert "secret_b64" not in str(payload)
|
|
assert "project_secret_b64" not in str(payload)
|
|
assert set(item["secret_kind"] for item in payload["project_secrets"]) == {"chat", "tabs", "gateway", "memory"}
|
|
|
|
for item in payload["project_secrets"]:
|
|
assert set(item) == {"secret_kind", "wrapped_key_bundle"}
|
|
assert_wrapped_record(mod, item["wrapped_key_bundle"], item["secret_kind"])
|
|
|
|
mod.DEFAULT_PROJECT_ID = PROJECT_ID
|
|
mod.require_unlocked = lambda operation="operation": None
|
|
mod.ACTIVE_MEMORY_SECRET_VERSIONS[PROJECT_ID] = 1
|
|
mod.MEMORY_SECRET_CACHE[(PROJECT_ID, "memory", 1, 1, 1, 1)] = bytes(range(32))
|
|
completion_calls = []
|
|
mod.request_project_write = lambda method, path, body, project_id, purpose, scope_type="project": completion_calls.append(
|
|
(method, path, body, project_id, purpose, scope_type)
|
|
) or {
|
|
"setup_request": {"id": "setup-1", "status": "completed"},
|
|
"project_secrets": [],
|
|
}
|
|
completed = mod.gateway_complete_vault_setup_request({"id": "setup-1", "project_id": PROJECT_ID})
|
|
assert completed["setup_request"]["status"] == "completed"
|
|
assert completion_calls[0][0] == "POST"
|
|
assert completion_calls[0][1] == f"/v1/projects/{PROJECT_ID}/vault/setup-requests/setup-1/complete"
|
|
assert completion_calls[0][5] == "encryption"
|
|
assert set(completion_calls[0][2]) == {"project_secrets"}
|
|
assert PROJECT_ID not in mod.ACTIVE_MEMORY_SECRET_VERSIONS
|
|
assert not mod.MEMORY_SECRET_CACHE
|
|
|
|
assert "project_vaults" not in mod.VAULT_SECRETS
|
|
assert "vault_master_secret_b64" not in str(mod.VAULT_SECRETS)
|
|
assert "secret_b64" not in str(mod.VAULT_SECRETS)
|
|
|
|
tools = {tool["name"]: tool for tool in mod.raw_tools()}
|
|
assert "nap_gateway_vault_ls" in tools
|
|
assert "nap_gateway_vault_process" in tools
|
|
assert "nap_gateway_vault_" + "setup_ls" not in tools
|
|
assert "nap_gateway_vault_" + "setup_process" not in tools
|
|
assert not [name for name in tools if name.startswith("napseer_")]
|
|
assert not [
|
|
tool for tool in mod.tools() if tool.get("napseer", {}).get("replaces")
|
|
]
|
|
assert "transient secret_b64" not in tools["nap_gateway_vault_process"]["description"]
|
|
assert "opaque client-wrapped key bundle records" in tools["nap_gateway_vault_process"]["description"]
|
|
assert "transient secret_b64" not in tools["nap_gateway_vault_secret_rotate"]["description"]
|
|
assert "opaque client-wrapped key bundle record" in tools["nap_gateway_vault_secret_rotate"]["description"]
|
|
|
|
filtered = mod.gateway_vault_secret_subset({
|
|
"gateway_encryption_key": base64.b64encode(bytes(range(32))).decode("ascii"),
|
|
"gateway_remote_enabled": True,
|
|
"gateway_default_command": "bash",
|
|
"project_vaults": {
|
|
PROJECT_ID: {
|
|
"vault_master_secret_b64": base64.b64encode(bytes(range(32))).decode("ascii"),
|
|
"project_secrets": {
|
|
"chat": {"secret_b64": base64.b64encode(bytes(range(32))).decode("ascii")},
|
|
},
|
|
},
|
|
},
|
|
})
|
|
assert filtered["gateway_default_command"] == "bash"
|
|
assert filtered["gateway_remote_enabled"] is True
|
|
assert "project_vaults" not in filtered
|
|
assert "vault_master_secret_b64" not in str(filtered)
|
|
assert "secret_b64" not in str(filtered)
|
|
|
|
local_before = {
|
|
"gateway_encryption_key": base64.b64encode(bytes(range(32))).decode("ascii"),
|
|
"gateway_remote_enabled": True,
|
|
"gateway_default_command": "bash",
|
|
}
|
|
mod.VAULT_SECRETS = dict(local_before)
|
|
assert len(mod.gateway_generate_project_vault_setup_payload(PROJECT_ID, "setup-2")["project_secrets"]) == 4
|
|
assert mod.VAULT_SECRETS == local_before
|
|
|
|
vector_plaintext = mod.generate_project_data_key_bundle(PROJECT_ID, ACCOUNT_ID, data_key_epoch=7)
|
|
vector_plaintext["bundle_id"] = "vector-bundle-001"
|
|
for index, kind in enumerate(mod.GATEWAY_PROJECT_SECRET_KINDS):
|
|
key_record = vector_plaintext["keys"][kind]["versions"]["7"]
|
|
key_record["key_b64"] = base64.b64encode(bytes([index + 1]) * 32).decode("ascii")
|
|
key_record["created_at"] = "2026-05-06T00:00:00Z"
|
|
vector_record = mod.wrapped_project_key_bundle_record(
|
|
PROJECT_ID,
|
|
ACCOUNT_ID,
|
|
"memory",
|
|
vector_plaintext,
|
|
wrapping_epoch=3,
|
|
bundle_version=7,
|
|
data_key_epoch=7,
|
|
salt=bytes(range(16)),
|
|
nonce=bytes(range(12)),
|
|
created_at="2026-05-06T00:00:00Z",
|
|
)
|
|
assert vector_record["aad_hash"] == "sha256:22f60004f8e873b4e3d9bd3b15ab0ba4de358ba74fc799255d0488596f33fa68"
|
|
assert vector_record["ciphertext_sha256"] == "sha256:5c59b781c3ad29843fb9f61efbd57ef6b19a963565a1e1f508bd18b20a6b6212"
|
|
assert mod.unwrap_project_key_bundle(vector_record) == vector_plaintext
|
|
|
|
calls = []
|
|
active_bundle = mod.wrapped_project_key_bundle_record(
|
|
PROJECT_ID,
|
|
ACCOUNT_ID,
|
|
"chat",
|
|
mod.generate_project_data_key_bundle(PROJECT_ID, ACCOUNT_ID, data_key_epoch=1),
|
|
wrapping_epoch=1,
|
|
bundle_version=1,
|
|
data_key_epoch=1,
|
|
)
|
|
|
|
def request_json(method, path, payload=None, **kwargs):
|
|
assert (method, path) == ("GET", f"/v1/projects/{PROJECT_ID}/vault/secrets/chat/active")
|
|
return {
|
|
"secret_kind": "chat",
|
|
"version": 1,
|
|
"wrapping_epoch": 1,
|
|
"bundle_version": 1,
|
|
"data_key_epoch": 1,
|
|
"wrapped_key_bundle": active_bundle,
|
|
}
|
|
|
|
mod.DEFAULT_PROJECT_ID = PROJECT_ID
|
|
mod.require_unlocked = lambda operation="operation": None
|
|
mod.request_json = request_json
|
|
mod.request_project_write = lambda method, path, body, project_id, purpose, scope_type="project": calls.append(
|
|
(method, path, body, project_id, purpose, scope_type)
|
|
) or {
|
|
"id": "secret-version-2",
|
|
"version": 2,
|
|
"secret_kind": "chat",
|
|
}
|
|
rotated = mod.gateway_rotate_project_vault_secret({"project_id": PROJECT_ID, "secret_kind": "chat"})
|
|
assert rotated["version"] == 2
|
|
assert not writes, "rotation must not persist project secret material locally"
|
|
assert calls[0][0] == "POST"
|
|
assert calls[0][1] == f"/v1/projects/{PROJECT_ID}/vault/secrets/chat/rotate"
|
|
assert calls[0][5] == "encryption"
|
|
assert set(calls[0][2]) == {
|
|
"previous_wrapping_epoch",
|
|
"previous_bundle_version",
|
|
"previous_data_key_epoch",
|
|
"previous_aad_hash",
|
|
"previous_ciphertext_sha256",
|
|
"wrapped_key_bundle",
|
|
}
|
|
assert calls[0][2]["previous_wrapping_epoch"] == 1
|
|
assert calls[0][2]["previous_bundle_version"] == 1
|
|
assert calls[0][2]["previous_data_key_epoch"] == 1
|
|
assert calls[0][2]["previous_aad_hash"] == active_bundle["aad_hash"]
|
|
assert calls[0][2]["previous_ciphertext_sha256"] == active_bundle["ciphertext_sha256"]
|
|
assert_wrapped_record(mod, calls[0][2]["wrapped_key_bundle"], "chat", data_key_epoch=2)
|
|
assert "secret_b64" not in str(calls[0][2])
|
|
assert mod.VAULT_SECRETS == local_before
|
|
|
|
print("ok: gateway vault setup payload smoke passed")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run()
|