473 lines
17 KiB
Python
473 lines
17 KiB
Python
import base64
|
|
import json
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
|
|
from memrelay.errors import AppError
|
|
from memrelay.models import CredentialMapping, VaultConnection
|
|
from memrelay.vault_service import VaultService
|
|
from tests.conftest import csrf_headers
|
|
|
|
|
|
class FakeBwRunner:
|
|
def __init__(self) -> None:
|
|
self.fail_sync = False
|
|
self.commands: list[list[str]] = []
|
|
self.items = [
|
|
{
|
|
"id": "item-1",
|
|
"name": "Git Service",
|
|
"login": {
|
|
"username": "developer",
|
|
"password": "git-password",
|
|
"totp": "JBSWY3DPEHPK3PXP",
|
|
"uris": [{"uri": "https://git.example.com"}],
|
|
},
|
|
"fields": [{"name": "api_token", "value": "token-value"}],
|
|
"notes": "private note",
|
|
},
|
|
{
|
|
"id": "item-2",
|
|
"name": "Git Mirror",
|
|
"login": {
|
|
"username": "mirror",
|
|
"password": "mirror-password",
|
|
"uris": [{"uri": "https://mirror.example.com"}],
|
|
},
|
|
"fields": [],
|
|
},
|
|
]
|
|
|
|
async def __call__(self, arguments: list[str], environment: dict[str, str]) -> str:
|
|
self.commands.append(arguments)
|
|
if arguments[0] in {"logout", "config"}:
|
|
return ""
|
|
if arguments[:2] == ["login", "--apikey"]:
|
|
assert environment == {
|
|
"BW_CLIENTID": "user.api-client-id",
|
|
"BW_CLIENTSECRET": "api-client-secret",
|
|
}
|
|
return ""
|
|
if arguments[0] in {"login", "unlock"}:
|
|
assert environment["BW_PASSWORD"] == "vault-password"
|
|
return "session-token"
|
|
if arguments[0] == "sync":
|
|
if self.fail_sync:
|
|
raise AppError("VAULT_UNAVAILABLE", "密码库服务不可达", 503)
|
|
return ""
|
|
if arguments[:2] == ["list", "items"]:
|
|
if "--search" not in arguments:
|
|
return json.dumps(self.items)
|
|
query = arguments[arguments.index("--search") + 1].casefold()
|
|
matches = [
|
|
item
|
|
for item in self.items
|
|
if query in item["name"].casefold()
|
|
or any(
|
|
query in str(uri.get("uri", "")).casefold()
|
|
for uri in (item.get("login") or {}).get("uris", [])
|
|
)
|
|
]
|
|
return json.dumps(matches)
|
|
if arguments[:2] == ["get", "item"]:
|
|
item_id = arguments[2]
|
|
return json.dumps(next(item for item in self.items if item["id"] == item_id))
|
|
if arguments[:2] == ["get", "totp"]:
|
|
return "123456"
|
|
if arguments[:2] == ["create", "item"]:
|
|
item = json.loads(base64.b64decode(arguments[2]).decode())
|
|
item["id"] = f"item-{len(self.items) + 1}"
|
|
item["revisionDate"] = "2026-08-03T00:00:00Z"
|
|
self.items.append(item)
|
|
return json.dumps(item)
|
|
if arguments[:2] == ["edit", "item"]:
|
|
item_id = arguments[2]
|
|
item = json.loads(base64.b64decode(arguments[3]).decode())
|
|
item["id"] = item_id
|
|
item["revisionDate"] = "2026-08-03T00:01:00Z"
|
|
index = next(
|
|
index for index, current in enumerate(self.items) if current["id"] == item_id
|
|
)
|
|
self.items[index] = item
|
|
return json.dumps(item)
|
|
if arguments[:2] == ["delete", "item"]:
|
|
item_id = arguments[2]
|
|
self.items = [item for item in self.items if item["id"] != item_id]
|
|
return ""
|
|
raise AssertionError(f"Unexpected bw command: {arguments}")
|
|
|
|
|
|
def install_fake_vault(client: TestClient, runner: FakeBwRunner) -> None:
|
|
client.app.state.vault = VaultService(
|
|
client.app.state.settings,
|
|
client.app.state.session_factory,
|
|
client.app.state.secret_box,
|
|
runner,
|
|
)
|
|
|
|
|
|
def configure_vault(client: TestClient) -> dict:
|
|
response = client.put(
|
|
"/api/v1/vault/connection",
|
|
json={
|
|
"server_url": "https://vault.example.com",
|
|
"account": "user@example.com",
|
|
"password": "vault-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def configure_vault_with_api_key(client: TestClient) -> dict:
|
|
response = client.put(
|
|
"/api/v1/vault/connection",
|
|
json={
|
|
"server_url": "https://vault.example.com",
|
|
"login_method": "api_key",
|
|
"client_id": "user.api-client-id",
|
|
"client_secret": "api-client-secret",
|
|
"password": "vault-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert response.status_code == 200, response.text
|
|
return response.json()
|
|
|
|
|
|
def test_configure_encrypt_sync_and_disconnect(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
runner = FakeBwRunner()
|
|
install_fake_vault(client, runner)
|
|
configured = configure_vault(client)
|
|
assert configured["status"] == "unlocked"
|
|
assert configured["last_sync_at"] is not None
|
|
|
|
with client.app.state.session_factory() as db:
|
|
connection = db.get(VaultConnection, 1)
|
|
assert connection is not None
|
|
assert "vault-password" not in connection.encrypted_password
|
|
assert "user@example.com" not in connection.encrypted_account
|
|
|
|
assert client.delete("/api/v1/vault/connection").status_code == 403
|
|
assert (
|
|
client.delete("/api/v1/vault/connection", headers=csrf_headers(client)).status_code == 204
|
|
)
|
|
assert client.get("/api/v1/vault/status").json()["configured"] is False
|
|
|
|
|
|
def test_configure_with_api_key_encrypts_and_unlocks(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
runner = FakeBwRunner()
|
|
install_fake_vault(client, runner)
|
|
|
|
configured = configure_vault_with_api_key(client)
|
|
|
|
assert configured["login_method"] == "api_key"
|
|
assert configured["account"] == "user.api-client-id"
|
|
assert ["login", "--apikey"] in runner.commands
|
|
assert ["unlock", "--passwordenv", "BW_PASSWORD", "--raw"] in runner.commands
|
|
with client.app.state.session_factory() as db:
|
|
connection = db.get(VaultConnection, 1)
|
|
assert connection is not None
|
|
assert connection.login_method == "api_key"
|
|
assert "api-client-secret" not in (connection.encrypted_client_secret or "")
|
|
assert "vault-password" not in connection.encrypted_password
|
|
|
|
|
|
def test_vault_login_fields_are_validated(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
install_fake_vault(client, FakeBwRunner())
|
|
|
|
display_name = client.put(
|
|
"/api/v1/vault/connection",
|
|
json={
|
|
"server_url": "https://vault.example.com",
|
|
"account": "display-name",
|
|
"password": "vault-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
missing_api_secret = client.put(
|
|
"/api/v1/vault/connection",
|
|
json={
|
|
"server_url": "https://vault.example.com",
|
|
"login_method": "api_key",
|
|
"client_id": "user.api-client-id",
|
|
"password": "vault-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
|
|
assert display_name.status_code == 422
|
|
assert missing_api_secret.status_code == 422
|
|
|
|
|
|
def test_search_resolve_secret_fields_totp_and_mapping(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
runner = FakeBwRunner()
|
|
install_fake_vault(client, runner)
|
|
configure_vault(client)
|
|
|
|
search = client.get("/api/v1/vault/secrets", params={"query": "Git"}).json()
|
|
assert len(search["results"]) == 2
|
|
assert "password" not in json.dumps(search)
|
|
|
|
ambiguous = client.post(
|
|
"/api/v1/vault/secrets/resolve",
|
|
json={"lookup_key": "Git"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert ambiguous.status_code == 409
|
|
assert ambiguous.json()["error"]["code"] == "SECRET_AMBIGUOUS"
|
|
selected = client.post(
|
|
"/api/v1/vault/secrets/resolve",
|
|
json={"lookup_key": "Git", "item_id": "item-1"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert selected.json()["id"] == "item-1"
|
|
|
|
password = client.post(
|
|
"/api/v1/vault/secrets/value",
|
|
json={"lookup_key": "Git", "field": "password"},
|
|
)
|
|
assert password.json()["value"] == "git-password"
|
|
assert password.headers["cache-control"] == "no-store"
|
|
custom = client.post(
|
|
"/api/v1/vault/secrets/value",
|
|
json={"lookup_key": "Git", "field": "api_token"},
|
|
)
|
|
assert custom.json()["value"] == "token-value"
|
|
totp = client.post("/api/v1/vault/secrets/totp", json={"lookup_key": "Git"})
|
|
assert totp.json()["value"] == "123456"
|
|
|
|
|
|
def test_list_get_create_update_and_delete_items(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
runner = FakeBwRunner()
|
|
install_fake_vault(client, runner)
|
|
configure_vault(client)
|
|
list_commands_after_sync = runner.commands.count(["list", "items"])
|
|
|
|
listed = client.get("/api/v1/vault/items")
|
|
assert listed.status_code == 200
|
|
assert len(listed.json()["results"]) == 2
|
|
assert "git-password" not in listed.text
|
|
|
|
detail = client.get("/api/v1/vault/items/item-1")
|
|
assert detail.status_code == 200
|
|
assert detail.headers["cache-control"] == "no-store"
|
|
assert detail.json()["password"] == "git-password"
|
|
assert runner.commands.count(["list", "items"]) == list_commands_after_sync
|
|
assert ["get", "item", "item-1"] not in runner.commands
|
|
|
|
payload = {
|
|
"name": "Package Registry",
|
|
"username": "builder",
|
|
"password": "registry-password",
|
|
"uris": ["https://packages.example.com"],
|
|
"notes": "Used by release jobs",
|
|
"totp": "otpauth://totp/example",
|
|
"fields": [
|
|
{"name": "access_token", "value": "token-value", "hidden": True},
|
|
{"name": "environment", "value": "production", "hidden": False},
|
|
],
|
|
"lookup_key": "registry",
|
|
}
|
|
assert client.post("/api/v1/vault/items", json=payload).status_code == 403
|
|
created = client.post("/api/v1/vault/items", json=payload, headers=csrf_headers(client))
|
|
assert created.status_code == 200, created.text
|
|
assert created.json()["created"] is True
|
|
item_id = created.json()["item"]["id"]
|
|
assert created.json()["item"]["fields"][0]["hidden"] is True
|
|
|
|
with client.app.state.session_factory() as db:
|
|
mapping = db.scalar(
|
|
select(CredentialMapping).where(CredentialMapping.lookup_key == "registry")
|
|
)
|
|
assert mapping is not None
|
|
assert mapping.vault_item_id == item_id
|
|
|
|
payload["password"] = "updated-password"
|
|
updated_by_uri = client.post("/api/v1/vault/items", json=payload, headers=csrf_headers(client))
|
|
assert updated_by_uri.status_code == 200
|
|
assert updated_by_uri.json()["created"] is False
|
|
assert updated_by_uri.json()["item"]["id"] == item_id
|
|
assert len(runner.items) == 3
|
|
|
|
payload["name"] = "Renamed Registry"
|
|
updated_by_id = client.put(
|
|
f"/api/v1/vault/items/{item_id}", json=payload, headers=csrf_headers(client)
|
|
)
|
|
assert updated_by_id.status_code == 200
|
|
assert updated_by_id.json()["item"]["name"] == "Renamed Registry"
|
|
|
|
assert client.delete(f"/api/v1/vault/items/{item_id}").status_code == 403
|
|
deleted = client.delete(f"/api/v1/vault/items/{item_id}", headers=csrf_headers(client))
|
|
assert deleted.status_code == 204
|
|
assert all(item["id"] != item_id for item in runner.items)
|
|
with client.app.state.session_factory() as db:
|
|
assert (
|
|
db.scalar(select(CredentialMapping).where(CredentialMapping.lookup_key == "registry"))
|
|
is None
|
|
)
|
|
|
|
|
|
def test_automatic_save_rejects_ambiguous_exact_matches(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
runner = FakeBwRunner()
|
|
runner.items[1]["name"] = "Git Service"
|
|
install_fake_vault(client, runner)
|
|
configure_vault(client)
|
|
|
|
response = client.post(
|
|
"/api/v1/vault/items",
|
|
json={"name": "Git Service", "password": "replacement"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
assert response.json()["error"]["code"] == "SECRET_AMBIGUOUS"
|
|
|
|
|
|
def test_automatic_save_uses_name_and_provided_identity_fields(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
runner = FakeBwRunner()
|
|
runner.items[1]["name"] = "Git Service"
|
|
install_fake_vault(client, runner)
|
|
configure_vault(client)
|
|
|
|
matched = client.post(
|
|
"/api/v1/vault/items",
|
|
json={
|
|
"name": "Git Service",
|
|
"username": "mirror",
|
|
"uris": ["https://mirror.example.com/"],
|
|
"password": "rotated-mirror-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert matched.status_code == 200, matched.text
|
|
assert matched.json()["created"] is False
|
|
assert matched.json()["item"]["id"] == "item-2"
|
|
|
|
distinct_account = client.post(
|
|
"/api/v1/vault/items",
|
|
json={
|
|
"name": "Git Service",
|
|
"username": "release-bot",
|
|
"uris": ["https://git.example.com"],
|
|
"password": "release-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert distinct_account.status_code == 200, distinct_account.text
|
|
assert distinct_account.json()["created"] is True
|
|
|
|
renamed_same_uri = client.post(
|
|
"/api/v1/vault/items",
|
|
json={
|
|
"name": "Git Production",
|
|
"username": "developer",
|
|
"uris": ["https://git.example.com"],
|
|
"password": "production-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert renamed_same_uri.status_code == 200, renamed_same_uri.text
|
|
assert renamed_same_uri.json()["created"] is True
|
|
|
|
|
|
def test_partial_automatic_update_preserves_unspecified_fields(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
runner = FakeBwRunner()
|
|
install_fake_vault(client, runner)
|
|
configure_vault(client)
|
|
|
|
response = client.post(
|
|
"/api/v1/vault/items",
|
|
json={"name": "Git Service", "password": "rotated-password"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
saved = response.json()["item"]
|
|
assert saved["password"] == "rotated-password"
|
|
assert saved["username"] == "developer"
|
|
assert saved["uris"] == ["https://git.example.com"]
|
|
assert saved["notes"] == "private note"
|
|
assert saved["totp"] == "JBSWY3DPEHPK3PXP"
|
|
assert saved["fields"] == [{"name": "api_token", "value": "token-value", "hidden": False}]
|
|
|
|
|
|
def test_unique_resolution_offline_cache_and_restart_unlock(initialized_client: TestClient) -> None:
|
|
client = initialized_client
|
|
runner = FakeBwRunner()
|
|
install_fake_vault(client, runner)
|
|
configure_vault(client)
|
|
unique = client.post(
|
|
"/api/v1/vault/secrets/resolve",
|
|
json={"lookup_key": "Mirror"},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert unique.json()["id"] == "item-2"
|
|
|
|
runner.fail_sync = True
|
|
failed = client.post("/api/v1/vault/sync", headers=csrf_headers(client))
|
|
assert failed.status_code == 503
|
|
assert client.get("/api/v1/vault/status").json()["status"] == "cached"
|
|
cached = client.get("/api/v1/vault/secrets", params={"query": "Mirror"}).json()
|
|
assert cached["cached"] is True
|
|
|
|
restarted_runner = FakeBwRunner()
|
|
restarted = VaultService(
|
|
client.app.state.settings,
|
|
client.app.state.session_factory,
|
|
client.app.state.secret_box,
|
|
restarted_runner,
|
|
)
|
|
client.app.state.vault = restarted
|
|
value = client.post(
|
|
"/api/v1/vault/secrets/value",
|
|
json={"lookup_key": "Mirror", "field": "username"},
|
|
)
|
|
assert value.json()["value"] == "mirror"
|
|
assert any(command[0] == "unlock" for command in restarted_runner.commands)
|
|
assert not any(command[0] == "config" for command in restarted_runner.commands)
|
|
|
|
|
|
def test_vault_accepts_private_http_and_rejects_invalid_schemes(
|
|
initialized_client: TestClient,
|
|
) -> None:
|
|
client = initialized_client
|
|
install_fake_vault(client, FakeBwRunner())
|
|
configured = client.put(
|
|
"/api/v1/vault/connection",
|
|
json={
|
|
"server_url": "http://vault.local",
|
|
"account": "user@example.com",
|
|
"password": "vault-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert configured.status_code == 200, configured.text
|
|
assert configured.json()["server_url"] == "http://vault.local"
|
|
|
|
invalid = client.put(
|
|
"/api/v1/vault/connection",
|
|
json={
|
|
"server_url": "ftp://vault.local",
|
|
"account": "user@example.com",
|
|
"password": "vault-password",
|
|
},
|
|
headers=csrf_headers(client),
|
|
)
|
|
assert invalid.status_code == 422
|
|
assert invalid.json()["error"]["code"] == "VAULT_URL_INVALID"
|