272 lines
11 KiB
Python
272 lines
11 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
|
|
|
|
@pytest.fixture
|
|
def api_client(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "secret_storage.db"
|
|
monkeypatch.setenv("APPOPS_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("APPOPS_DATA_DIR", str(tmp_path / "data"))
|
|
monkeypatch.setenv("APPOPS_ENV", "development")
|
|
monkeypatch.setenv("APPOPS_SECRET_KEY", "test-secret-storage-master-key")
|
|
monkeypatch.delenv("APPOPS_SECRET_KEY_FILE", raising=False)
|
|
for name in [
|
|
"PORTAINER_URL",
|
|
"PORTAINER_API_KEY",
|
|
"PORTAINER_ENDPOINT_ID",
|
|
"UNRAID_URL",
|
|
"UNRAID_API_KEY",
|
|
"SONARR_URL",
|
|
"SONARR_API_KEY",
|
|
"WIDEFROG_URL",
|
|
"WIDEFROG_API_KEY",
|
|
]:
|
|
monkeypatch.delenv(name, raising=False)
|
|
|
|
from backend.app.main import app
|
|
from backend.app.core.config import settings
|
|
from backend.app.db import session
|
|
from backend.app.db.init_db import init_db
|
|
|
|
previous = {
|
|
"db_path": settings.db_path,
|
|
"data_dir": settings.data_dir,
|
|
"environment": settings.environment,
|
|
"secret_key": getattr(settings, "secret_key", None),
|
|
"secret_key_source": getattr(settings, "secret_key_source", None),
|
|
"apis": dict(settings.apis),
|
|
}
|
|
settings.db_path = str(db_path)
|
|
settings.data_dir = str(tmp_path / "data")
|
|
settings.environment = "development"
|
|
settings.secret_key = "test-secret-storage-master-key"
|
|
settings.secret_key_source = "env"
|
|
settings.secret_key_file = None
|
|
for key in previous["apis"]:
|
|
if key.endswith("_API_KEY") or key in {"PORTAINER_URL", "PORTAINER_ENDPOINT_ID", "UNRAID_URL", "SONARR_URL", "WIDEFROG_URL"}:
|
|
settings.apis[key] = None
|
|
|
|
engine = create_engine(f"sqlite:///{Path(db_path)}", connect_args={"check_same_thread": False})
|
|
session.engine = engine
|
|
SQLModel.metadata.drop_all(engine)
|
|
SQLModel.metadata.create_all(engine)
|
|
init_db()
|
|
|
|
with TestClient(app) as client:
|
|
yield client
|
|
|
|
settings.db_path = previous["db_path"]
|
|
settings.data_dir = previous["data_dir"]
|
|
settings.environment = previous["environment"]
|
|
settings.secret_key = previous["secret_key"]
|
|
settings.secret_key_source = previous["secret_key_source"]
|
|
settings.apis = previous["apis"]
|
|
|
|
|
|
def test_secret_store_encrypts_at_rest_and_decrypts_internally(api_client):
|
|
from backend.app.db.models import Secret
|
|
from backend.app.db.session import engine
|
|
from backend.app.services.secret_store import get_secret, has_secret, list_secret_status, set_secret
|
|
|
|
secret_value = "PORTAINER_SECRET_AT_REST_VALUE"
|
|
with Session(engine) as session:
|
|
set_secret(session, "PORTAINER_API_KEY", secret_value, used_by=["portainer"])
|
|
row = session.get(Secret, "PORTAINER_API_KEY")
|
|
|
|
assert row is not None
|
|
assert row.encrypted_value
|
|
assert row.encrypted_value != secret_value
|
|
assert secret_value not in row.encrypted_value
|
|
assert has_secret(session, "PORTAINER_API_KEY") is True
|
|
assert get_secret(session, "PORTAINER_API_KEY") == secret_value
|
|
|
|
statuses = list_secret_status(session)
|
|
portainer_status = next(item for item in statuses if item["secretRef"] == "PORTAINER_API_KEY")
|
|
assert portainer_status["hasSecret"] is True
|
|
assert portainer_status["status"] == "configured"
|
|
assert "value" not in portainer_status
|
|
assert secret_value not in str(statuses)
|
|
|
|
|
|
def test_connector_settings_store_secret_encrypted_and_keep_response_safe(api_client):
|
|
from backend.app.db.models import AppSetting, Secret
|
|
from backend.app.db.session import engine
|
|
|
|
secret_value = "PORTAINER_PATCH_SECRET"
|
|
response = api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={
|
|
"url": "http://198.18.10.150:9200",
|
|
"endpointId": "1",
|
|
"verifyTls": False,
|
|
"apiKey": secret_value,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["connectorId"] == "portainer"
|
|
assert payload["configured"] is True
|
|
assert payload["hasSecret"] is True
|
|
assert payload["secretRef"] == "PORTAINER_API_KEY"
|
|
assert payload["secretStatus"]["status"] == "configured"
|
|
assert payload["safeConfig"] == {
|
|
"url": "http://198.18.10.150:9200",
|
|
"endpointId": "1",
|
|
"verifyTls": False,
|
|
}
|
|
assert secret_value not in str(payload)
|
|
|
|
with Session(engine) as session:
|
|
row = session.get(Secret, "PORTAINER_API_KEY")
|
|
assert row is not None
|
|
assert secret_value not in row.encrypted_value
|
|
|
|
settings_rows = session.exec(select(AppSetting)).all()
|
|
assert secret_value not in str([item.value_json for item in settings_rows])
|
|
|
|
listed = api_client.get("/api/settings/connectors").json()
|
|
assert secret_value not in str(listed)
|
|
assert next(item for item in listed if item["connectorId"] == "portainer")["hasSecret"] is True
|
|
|
|
|
|
def test_runtime_exports_and_secrets_endpoint_never_show_secret_values(api_client):
|
|
secret_values = {
|
|
"PORTAINER_API_KEY": "PORTAINER_EXPORT_SECRET",
|
|
"UNRAID_API_KEY": "UNRAID_EXPORT_SECRET",
|
|
"SONARR_API_KEY": "SONARR_EXPORT_SECRET",
|
|
"WIDEFROG_API_KEY": "WIDEFROG_EXPORT_SECRET",
|
|
}
|
|
api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={"url": "http://portainer.local", "endpointId": "1", "verifyTls": False, "apiKey": secret_values["PORTAINER_API_KEY"]},
|
|
)
|
|
api_client.patch("/api/settings/connectors/unraid", json={"url": "http://unraid.local", "apiKey": secret_values["UNRAID_API_KEY"]})
|
|
api_client.patch("/api/settings/connectors/sonarr", json={"url": "http://sonarr.local", "apiKey": secret_values["SONARR_API_KEY"]})
|
|
api_client.patch("/api/settings/connectors/widefrog", json={"url": "http://widefrog.local", "apiKey": secret_values["WIDEFROG_API_KEY"]})
|
|
|
|
for path in [
|
|
"/api/settings/connectors",
|
|
"/api/settings/secrets",
|
|
"/api/setup/status",
|
|
"/api/runtime/info",
|
|
"/api/admin/export/config",
|
|
"/api/admin/export/policies",
|
|
"/api/admin/export/inventory",
|
|
"/api/admin/export/audit-summary",
|
|
]:
|
|
response = api_client.get(path)
|
|
assert response.status_code == 200
|
|
body = response.text
|
|
for secret in secret_values.values():
|
|
assert secret not in body
|
|
assert "test-secret-storage-master-key" not in body
|
|
|
|
secrets_payload = api_client.get("/api/settings/secrets").json()
|
|
refs = {item["secretRef"] for item in secrets_payload["secrets"]}
|
|
assert {"PORTAINER_API_KEY", "UNRAID_API_KEY", "SONARR_API_KEY", "WIDEFROG_API_KEY"}.issubset(refs)
|
|
assert secrets_payload["storage"]["encryptedAtRest"] is True
|
|
assert secrets_payload["storage"]["keySource"] == "env"
|
|
|
|
|
|
def test_setup_status_does_not_touch_secret_last_used(api_client):
|
|
from backend.app.db.models import Secret
|
|
from backend.app.db.session import engine
|
|
|
|
api_client.patch(
|
|
"/api/settings/connectors/portainer",
|
|
json={
|
|
"url": "http://portainer.local",
|
|
"endpointId": "1",
|
|
"verifyTls": False,
|
|
"apiKey": "PORTAINER_READONLY_STATUS_SECRET",
|
|
},
|
|
)
|
|
|
|
with Session(engine) as session:
|
|
row = session.get(Secret, "PORTAINER_API_KEY")
|
|
assert row is not None
|
|
row.last_used_at = None
|
|
session.add(row)
|
|
session.commit()
|
|
|
|
response = api_client.get("/api/setup/status")
|
|
|
|
assert response.status_code == 200
|
|
with Session(engine) as session:
|
|
row = session.get(Secret, "PORTAINER_API_KEY")
|
|
assert row is not None
|
|
assert row.last_used_at is None
|
|
|
|
|
|
def test_legacy_plaintext_connector_secret_is_migrated(api_client):
|
|
from backend.app.core.config import settings
|
|
from backend.app.db.models import AppSetting, Secret
|
|
from backend.app.db.session import engine
|
|
from backend.app.services.setup_service import apply_stored_connector_settings, serialize_connector_settings
|
|
|
|
legacy_secret = "LEGACY_PORTAINER_PLAINTEXT"
|
|
with Session(engine) as session:
|
|
session.add(AppSetting(key="connector:portainer:config", value_json={"url": "http://legacy.local", "endpointId": "1"}, is_secret=False))
|
|
session.add(AppSetting(key="connector:portainer:secret", value_json={"value": legacy_secret}, is_secret=True))
|
|
session.commit()
|
|
|
|
apply_stored_connector_settings(session)
|
|
payload = serialize_connector_settings(session, "portainer")
|
|
legacy_row = session.get(AppSetting, "connector:portainer:secret")
|
|
secret_row = session.get(Secret, "PORTAINER_API_KEY")
|
|
|
|
assert payload["hasSecret"] is True
|
|
assert payload["secretRef"] == "PORTAINER_API_KEY"
|
|
assert settings.apis["PORTAINER_API_KEY"] == legacy_secret
|
|
assert secret_row is not None
|
|
assert legacy_secret not in secret_row.encrypted_value
|
|
assert legacy_secret not in str(legacy_row.value_json)
|
|
assert legacy_row.value_json["secretRef"] == "PORTAINER_API_KEY"
|
|
|
|
response = api_client.get("/api/settings/connectors")
|
|
assert response.status_code == 200
|
|
assert legacy_secret not in response.text
|
|
|
|
|
|
def test_secret_redaction_masks_common_tokens_and_known_values(api_client):
|
|
from backend.app.db.session import engine
|
|
from backend.app.services.secret_store import redact_text, set_secret
|
|
|
|
with Session(engine) as session:
|
|
set_secret(session, "SONARR_API_KEY", "KNOWN_SONARR_SECRET", used_by=["sonarr"])
|
|
message = (
|
|
"Bearer abcdefghijklmnop x-api-key: XAPISECRET "
|
|
"apiKey=INLINESECRET password=hunter2 known=KNOWN_SONARR_SECRET"
|
|
)
|
|
redacted = redact_text(message, session=session)
|
|
|
|
assert "abcdefghijklmnop" not in redacted
|
|
assert "XAPISECRET" not in redacted
|
|
assert "INLINESECRET" not in redacted
|
|
assert "hunter2" not in redacted
|
|
assert "KNOWN_SONARR_SECRET" not in redacted
|
|
assert "[REDACTED]" in redacted
|
|
|
|
|
|
def test_production_missing_secret_key_reports_warning(api_client):
|
|
from backend.app.core.config import settings
|
|
from backend.app.services.secret_store import secret_storage_status
|
|
|
|
previous = (settings.environment, settings.secret_key, settings.secret_key_source)
|
|
settings.environment = "production"
|
|
settings.secret_key = None
|
|
settings.secret_key_source = "missing"
|
|
|
|
status = secret_storage_status()
|
|
|
|
assert status["productionReady"] is False
|
|
assert status["keySource"] == "missing"
|
|
assert status["warnings"]
|
|
assert "APPOPS_SECRET_KEY" in status["warnings"][0]
|
|
|
|
settings.environment, settings.secret_key, settings.secret_key_source = previous
|