278 lines
12 KiB
Python
278 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import create_engine, inspect
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from govoplan_access.backend.db import models as access_models # noqa: F401 - resolve Files user foreign keys
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
|
from govoplan_core.db.base import Base
|
|
from govoplan_core.security.secrets import encrypt_secret
|
|
from govoplan_files.backend.db.models import FileConnectorCredential, FileConnectorProfile
|
|
from govoplan_files.backend.router import deactivate_connector_credential, deactivate_connector_profile
|
|
|
|
|
|
class Principal:
|
|
tenant_id = "tenant-1"
|
|
user = SimpleNamespace(id="user-1")
|
|
api_key = None
|
|
|
|
def has(self, scope: str) -> bool:
|
|
return scope == "files:file:admin"
|
|
|
|
|
|
class ConnectorCredentialDeletionTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
self.engine,
|
|
tables=[
|
|
FileConnectorCredential.__table__,
|
|
FileConnectorProfile.__table__,
|
|
ChangeSequenceEntry.__table__,
|
|
],
|
|
)
|
|
self.session = sessionmaker(bind=self.engine)()
|
|
self.principal = Principal()
|
|
|
|
def tearDown(self) -> None:
|
|
self.session.close()
|
|
Base.metadata.drop_all(bind=self.engine)
|
|
self.engine.dispose()
|
|
|
|
def test_delete_credential_scrubs_material_audits_and_disables_dependents(self) -> None:
|
|
credential = self._credential()
|
|
profile = self._profile(credential_profile_id=credential.id)
|
|
self.session.add_all([credential, profile])
|
|
self.session.commit()
|
|
|
|
with patch(
|
|
"govoplan_files.backend.storage.connector_credential_deletion.audit_event"
|
|
) as audit:
|
|
response = deactivate_connector_credential(
|
|
credential.id,
|
|
session=self.session,
|
|
principal=self.principal, # type: ignore[arg-type]
|
|
)
|
|
|
|
self.assertFalse(response.enabled)
|
|
self.assertFalse(response.credentials_configured)
|
|
self.assertIsNone(response.credential_secret_source)
|
|
self.session.refresh(credential)
|
|
self.session.refresh(profile)
|
|
self._assert_credential_scrubbed(credential)
|
|
self._assert_profile_scrubbed(profile)
|
|
|
|
calls = {call.kwargs["action"]: call.kwargs for call in audit.call_args_list}
|
|
self.assertEqual(
|
|
{"files.connector_profile_deleted", "files.connector_credential_deleted"},
|
|
set(calls),
|
|
)
|
|
credential_audit = calls["files.connector_credential_deleted"]
|
|
self.assertEqual("encrypted_database", credential_audit["details"]["storage_backend"])
|
|
self.assertEqual(["password", "token"], credential_audit["details"]["deleted_secret_kinds"])
|
|
self.assertEqual(1, credential_audit["details"]["affected_profile_count"])
|
|
self.assertNotIn("do-not-audit", repr(audit.call_args_list))
|
|
|
|
changes = self.session.query(ChangeSequenceEntry).order_by(ChangeSequenceEntry.id.asc()).all()
|
|
self.assertEqual([credential.id, profile.id], [change.resource_id for change in changes])
|
|
self.assertEqual(["deleted", "updated"], [change.operation for change in changes])
|
|
|
|
def test_delete_profile_scrubs_inline_credentials_and_audits(self) -> None:
|
|
profile = self._profile(credential_profile_id="shared-credential")
|
|
self.session.add(profile)
|
|
self.session.commit()
|
|
|
|
with patch(
|
|
"govoplan_files.backend.storage.connector_credential_deletion.audit_event"
|
|
) as audit:
|
|
response = deactivate_connector_profile(
|
|
profile.id,
|
|
session=self.session,
|
|
principal=self.principal, # type: ignore[arg-type]
|
|
)
|
|
|
|
self.assertFalse(response.enabled)
|
|
self.assertFalse(response.credentials_configured)
|
|
self.assertIsNone(response.credential_profile_id)
|
|
self.session.refresh(profile)
|
|
self._assert_profile_scrubbed(profile)
|
|
audit.assert_called_once()
|
|
call = audit.call_args.kwargs
|
|
self.assertEqual("files.connector_profile_deleted", call["action"])
|
|
self.assertEqual("api_delete", call["details"]["deletion_reason"])
|
|
self.assertIn("credential_profile", call["details"]["removed_reference_kinds"])
|
|
self.assertNotIn("do-not-audit", repr(call))
|
|
|
|
def test_delete_rolls_back_when_audit_fails(self) -> None:
|
|
credential = self._credential()
|
|
self.session.add(credential)
|
|
self.session.commit()
|
|
original_password = credential.password_encrypted
|
|
original_token = credential.token_encrypted
|
|
|
|
with patch(
|
|
"govoplan_files.backend.storage.connector_credential_deletion.audit_event",
|
|
side_effect=RuntimeError("audit unavailable"),
|
|
), self.assertRaisesRegex(RuntimeError, "audit unavailable"):
|
|
deactivate_connector_credential(
|
|
credential.id,
|
|
session=self.session,
|
|
principal=self.principal, # type: ignore[arg-type]
|
|
)
|
|
|
|
persisted = self.session.get(FileConnectorCredential, credential.id)
|
|
assert persisted is not None
|
|
self.assertTrue(persisted.enabled)
|
|
self.assertEqual(original_password, persisted.password_encrypted)
|
|
self.assertEqual(original_token, persisted.token_encrypted)
|
|
self.assertEqual(0, self.session.query(ChangeSequenceEntry).count())
|
|
|
|
def test_delete_detaches_unowned_legacy_secret_reference_without_claiming_provider_deletion(self) -> None:
|
|
credential = self._credential(secret_ref="vault:tenant-1:files:credential")
|
|
self.session.add(credential)
|
|
self.session.commit()
|
|
|
|
with patch(
|
|
"govoplan_files.backend.storage.connector_credential_deletion.audit_event"
|
|
) as audit:
|
|
response = deactivate_connector_credential(
|
|
credential.id,
|
|
session=self.session,
|
|
principal=self.principal, # type: ignore[arg-type]
|
|
)
|
|
|
|
self.assertFalse(response.enabled)
|
|
persisted = self.session.get(FileConnectorCredential, credential.id)
|
|
assert persisted is not None
|
|
self._assert_credential_scrubbed(persisted)
|
|
audit.assert_called_once()
|
|
details = audit.call_args.kwargs["details"]
|
|
self.assertEqual("unowned_external_reference_detached", details["storage_backend"])
|
|
self.assertIn("unowned_external_secret_ref", details["removed_reference_kinds"])
|
|
self.assertNotIn("vault:tenant-1:files:credential", repr(audit.call_args))
|
|
|
|
def test_retirement_scrubs_and_audits_before_tables_are_dropped(self) -> None:
|
|
from govoplan_files.backend.manifest import manifest
|
|
|
|
self.session.add_all([self._credential(), self._profile()])
|
|
self.session.commit()
|
|
retirement_provider = manifest.migration_spec.retirement_provider
|
|
assert retirement_provider is not None
|
|
plan = retirement_provider(self.session, "files")
|
|
assert plan.destroy_data_executor is not None
|
|
|
|
with patch(
|
|
"govoplan_files.backend.storage.connector_credential_deletion.audit_event"
|
|
) as audit:
|
|
plan.destroy_data_executor(self.session, "files")
|
|
# The retirement executor deliberately enlists secret scrubbing, audit
|
|
# writes, and DDL in the caller-owned transaction. Commit that unit of
|
|
# work before inspecting through a separate Engine connection.
|
|
self.session.commit()
|
|
|
|
self.assertEqual(2, audit.call_count)
|
|
self.assertEqual(
|
|
{"module_data_retired"},
|
|
{call.kwargs["details"]["deletion_reason"] for call in audit.call_args_list},
|
|
)
|
|
self.assertNotIn("do-not-audit", repr(audit.call_args_list))
|
|
self.assertFalse(inspect(self.engine).has_table(FileConnectorCredential.__tablename__))
|
|
self.assertFalse(inspect(self.engine).has_table(FileConnectorProfile.__tablename__))
|
|
|
|
def test_retirement_detaches_unowned_external_reference_before_drop(self) -> None:
|
|
from govoplan_files.backend.manifest import manifest
|
|
|
|
self.session.add(self._credential(secret_ref="vault:tenant-1:files:credential"))
|
|
self.session.commit()
|
|
retirement_provider = manifest.migration_spec.retirement_provider
|
|
assert retirement_provider is not None
|
|
plan = retirement_provider(self.session, "files")
|
|
assert plan.destroy_data_executor is not None
|
|
|
|
with patch(
|
|
"govoplan_files.backend.storage.connector_credential_deletion.audit_event"
|
|
) as audit:
|
|
plan.destroy_data_executor(self.session, "files")
|
|
self.session.commit()
|
|
|
|
audit.assert_called_once()
|
|
details = audit.call_args.kwargs["details"]
|
|
self.assertIn("unowned_external_secret_ref", details["removed_reference_kinds"])
|
|
self.assertNotIn("vault:tenant-1:files:credential", repr(audit.call_args))
|
|
self.assertFalse(inspect(self.engine).has_table(FileConnectorCredential.__tablename__))
|
|
self.assertFalse(inspect(self.engine).has_table(FileConnectorProfile.__tablename__))
|
|
|
|
@staticmethod
|
|
def _credential(*, secret_ref: str | None = None) -> FileConnectorCredential:
|
|
return FileConnectorCredential(
|
|
id="credential-1",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
label="Credential",
|
|
provider="webdav",
|
|
enabled=True,
|
|
credential_mode="basic",
|
|
username="connector-user",
|
|
password_encrypted=encrypt_secret("do-not-audit-password"),
|
|
token_encrypted=encrypt_secret("do-not-audit-token"),
|
|
password_env="DEPLOYMENT_PASSWORD",
|
|
token_env="DEPLOYMENT_TOKEN",
|
|
secret_ref=secret_ref,
|
|
policy={},
|
|
metadata_={"private_hint": "do-not-audit-metadata"},
|
|
)
|
|
|
|
@staticmethod
|
|
def _profile(*, credential_profile_id: str | None = None) -> FileConnectorProfile:
|
|
return FileConnectorProfile(
|
|
id="profile-1",
|
|
tenant_id="tenant-1",
|
|
scope_type="tenant",
|
|
scope_id="tenant-1",
|
|
label="Profile",
|
|
provider="webdav",
|
|
endpoint_url="https://dav.example.test",
|
|
enabled=True,
|
|
credential_profile_id=credential_profile_id,
|
|
credential_mode="basic",
|
|
username="connector-user",
|
|
password_encrypted=encrypt_secret("do-not-audit-profile-password"),
|
|
token_encrypted=encrypt_secret("do-not-audit-profile-token"),
|
|
password_env="DEPLOYMENT_PROFILE_PASSWORD",
|
|
token_env="DEPLOYMENT_PROFILE_TOKEN",
|
|
policy={},
|
|
metadata_={"private_hint": "do-not-audit-profile-metadata"},
|
|
)
|
|
|
|
def _assert_credential_scrubbed(self, row: FileConnectorCredential) -> None:
|
|
self.assertFalse(row.enabled)
|
|
self.assertEqual("none", row.credential_mode)
|
|
self.assertIsNone(row.username)
|
|
self.assertIsNone(row.password_encrypted)
|
|
self.assertIsNone(row.token_encrypted)
|
|
self.assertIsNone(row.password_env)
|
|
self.assertIsNone(row.token_env)
|
|
self.assertIsNone(row.secret_ref)
|
|
self.assertEqual({}, row.metadata_)
|
|
|
|
def _assert_profile_scrubbed(self, row: FileConnectorProfile) -> None:
|
|
self.assertFalse(row.enabled)
|
|
self.assertIsNone(row.credential_profile_id)
|
|
self.assertEqual("none", row.credential_mode)
|
|
self.assertIsNone(row.username)
|
|
self.assertIsNone(row.password_encrypted)
|
|
self.assertIsNone(row.token_encrypted)
|
|
self.assertIsNone(row.password_env)
|
|
self.assertIsNone(row.token_env)
|
|
self.assertIsNone(row.secret_ref)
|
|
self.assertEqual({}, row.metadata_)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|