GovOPlaN/govoplan-mail
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

security: scrub credentials when Mail profiles are deactivated

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-21 17:12:49 +02:00
parent 88d5012dea
commit 4a9201aeea
2 changed files with 177 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/govoplan_mail/backend/router.py
View File

@@ -53,6 +53,7 @@ from govoplan_mail.backend.mailbox_index import (
from govoplan_mail.backend.mail_profiles import (
MailProfileError,
create_mail_server_profile,
delete_mail_profile_credentials,
effective_mail_profile_policy_for_scope,
get_mail_profile_policy,
get_mail_server_profile,
@@ -700,7 +701,7 @@ def create_profile(
resource_id=profile.id,
operation="created",
principal=principal,
tenant_id=profile.tenant_id or principal.tenant_id,
tenant_id=profile.tenant_id,
payload={"scope_type": profile.scope_type, "scope_id": profile.scope_id},
)
session.commit()
@@ -751,6 +752,7 @@ def update_profile(
profile,
tenant_id=principal.tenant_id,
user_id=principal.user.id,
api_key_id=principal.api_key_id,
name=payload.name,
slug=payload.slug,
description=payload.description if "description" in payload.model_fields_set else None,
@@ -766,7 +768,7 @@ def update_profile(
resource_id=profile.id,
operation="updated",
principal=principal,
tenant_id=profile.tenant_id or principal.tenant_id,
tenant_id=profile.tenant_id,
payload={"scope_type": profile.scope_type, "scope_id": profile.scope_id},
)
session.commit()
@@ -775,6 +777,9 @@ def update_profile(
except MailProfileError as exc:
session.rollback()
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
except Exception:
session.rollback()
raise
@router.delete("/profiles/{profile_id}", response_model=MailServerProfileResponse)
@@ -786,8 +791,18 @@ def deactivate_profile(
try:
profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id)
_require_profile_write_scope(principal, profile.scope_type or "tenant")
_require_profile_credentials_scope(principal, profile.scope_type or "tenant")
was_active = bool(profile.is_active)
profile.is_active = False
deleted_protocols = delete_mail_profile_credentials(
session,
profile=profile,
deletion_reason="profile_deactivated",
user_id=principal.user.id,
api_key_id=principal.api_key_id,
)
session.add(profile)
if was_active or deleted_protocols:
_record_mail_change(
session,
collection=MAIL_PROFILES_COLLECTION,
@@ -795,8 +810,13 @@ def deactivate_profile(
resource_id=profile.id,
operation="deleted",
principal=principal,
tenant_id=profile.tenant_id or principal.tenant_id,
payload={"scope_type": profile.scope_type, "scope_id": profile.scope_id},
tenant_id=profile.tenant_id,
payload={
"scope_type": profile.scope_type,
"scope_id": profile.scope_id,
"credentials_deleted": bool(deleted_protocols),
"deleted_credential_protocols": list(deleted_protocols),
},
)
session.commit()
session.refresh(profile)
@@ -804,6 +824,9 @@ def deactivate_profile(
except MailProfileError as exc:
session.rollback()
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
except Exception:
session.rollback()
raise
@router.get("/policies/{scope_type}", response_model=MailProfilePolicyResponse)

143
tests/test_router_profile_deletion.py Normal file
View File

@@ -0,0 +1,143 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from unittest.mock import patch
from govoplan_mail.backend.router import create_profile, deactivate_profile, update_profile
from govoplan_mail.backend.schemas import MailServerProfileCreateRequest, MailServerProfileUpdateRequest
class _Session:
def __init__(self) -> None:
self.commits = 0
self.rollbacks = 0
def add(self, _value) -> None:
return None
def commit(self) -> None:
self.commits += 1
def rollback(self) -> None:
self.rollbacks += 1
def refresh(self, _value) -> None:
return None
class MailProfileDeletionRouteTests(unittest.TestCase):
def setUp(self) -> None:
self.principal = SimpleNamespace(
tenant_id="tenant-1",
user=SimpleNamespace(id="user-1"),
api_key_id=None,
)
self.profile = SimpleNamespace(
id="profile-1",
tenant_id="tenant-1",
scope_type="tenant",
scope_id="tenant-1",
is_active=False,
)
def test_repeated_delete_does_not_emit_a_false_change(self) -> None:
session = _Session()
with (
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile),
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
patch("govoplan_mail.backend.router._record_mail_change") as record_change,
patch("govoplan_mail.backend.router._profile_response", return_value=self.profile),
):
result = deactivate_profile("profile-1", principal=self.principal, session=session)
self.assertIs(result, self.profile)
self.assertEqual(session.commits, 1)
self.assertEqual(session.rollbacks, 0)
record_change.assert_not_called()
def test_change_feed_reports_whether_credentials_were_deleted(self) -> None:
self.profile.is_active = True
session = _Session()
with (
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile),
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
patch("govoplan_mail.backend.router._record_mail_change") as record_change,
patch("govoplan_mail.backend.router._profile_response", return_value=self.profile),
):
deactivate_profile("profile-1", principal=self.principal, session=session)
self.assertFalse(record_change.call_args.kwargs["payload"]["credentials_deleted"])
self.assertEqual(record_change.call_args.kwargs["payload"]["deleted_credential_protocols"], [])
def test_generic_secret_deletion_failure_rolls_back(self) -> None:
self.profile.is_active = True
session = _Session()
with (
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile),
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", side_effect=RuntimeError("audit unavailable")),
):
with self.assertRaisesRegex(RuntimeError, "audit unavailable"):
deactivate_profile("profile-1", principal=self.principal, session=session)
self.assertEqual(session.commits, 0)
self.assertEqual(session.rollbacks, 1)
def test_system_profile_changes_are_instance_wide_in_the_change_feed(self) -> None:
system_profile = SimpleNamespace(
id="profile-system",
tenant_id=None,
scope_type="system",
scope_id=None,
is_active=True,
smtp_config={"host": "smtp.example.test"},
imap_config=None,
)
session = _Session()
create_payload = MailServerProfileCreateRequest.model_validate(
{
"name": "System profile",
"scope_type": "system",
"smtp": {"host": "smtp.example.test"},
}
)
with (
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router.create_mail_server_profile", return_value=system_profile),
patch("govoplan_mail.backend.router._record_mail_change") as create_change,
patch("govoplan_mail.backend.router._profile_response", return_value=system_profile),
):
create_profile(create_payload, principal=self.principal, session=session)
self.assertIsNone(create_change.call_args.kwargs["tenant_id"])
update_payload = MailServerProfileUpdateRequest(name="Renamed system profile")
with (
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=system_profile),
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router.update_mail_server_profile", return_value=system_profile),
patch("govoplan_mail.backend.router._record_mail_change") as update_change,
patch("govoplan_mail.backend.router._profile_response", return_value=system_profile),
):
update_profile("profile-system", update_payload, principal=self.principal, session=session)
self.assertIsNone(update_change.call_args.kwargs["tenant_id"])
with (
patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=system_profile),
patch("govoplan_mail.backend.router._require_profile_write_scope"),
patch("govoplan_mail.backend.router._require_profile_credentials_scope"),
patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()),
patch("govoplan_mail.backend.router._record_mail_change") as delete_change,
patch("govoplan_mail.backend.router._profile_response", return_value=system_profile),
):
deactivate_profile("profile-system", principal=self.principal, session=session)
self.assertIsNone(delete_change.call_args.kwargs["tenant_id"])
if __name__ == "__main__":
unittest.main()