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

Harden mail profile validation

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-11 18:37:51 +02:00
parent 5371cb3ad4
commit 4865de5007
2 changed files with 45 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/govoplan_mail/backend/mail_profiles.py
View File

@@ -285,11 +285,15 @@ def _legacy_tenant_settings(session: Session, tenant_id: str) -> dict[str, Any]
return _json_object(row[0]) if row else None
def _legacy_subject_policy(session: Session, *, table_name: str, tenant_id: str, scope_id: str) -> dict[str, Any] | None:
if table_name not in {"access_users", "access_groups"}:
def _legacy_subject_policy(session: Session, *, subject_type: str, tenant_id: str, scope_id: str) -> dict[str, Any] | None:
if subject_type == "user":
statement = text("select mail_profile_policy from access_users where id = :scope_id and tenant_id = :tenant_id")
elif subject_type == "group":
statement = text("select mail_profile_policy from access_groups where id = :scope_id and tenant_id = :tenant_id")
else:
raise MailProfileError("Unsupported mail policy subject")
row = session.execute(
text(f"select mail_profile_policy from {table_name} where id = :scope_id and tenant_id = :tenant_id"),
statement,
{"scope_id": scope_id, "tenant_id": tenant_id},
).first()
return _json_object(row[0]) if row else None
@@ -304,7 +308,7 @@ def _user_exists(session: Session, *, tenant_id: str, user_id: str) -> bool:
if directory is not None:
user = directory.get_user(user_id)
return bool(user and user.tenant_id == tenant_id and user.status == "active")
return _legacy_subject_policy(session, table_name="access_users", tenant_id=tenant_id, scope_id=user_id) is not None
return _legacy_subject_policy(session, subject_type="user", tenant_id=tenant_id, scope_id=user_id) is not None
def _group_exists(session: Session, *, tenant_id: str, group_id: str) -> bool:
@@ -312,7 +316,7 @@ def _group_exists(session: Session, *, tenant_id: str, group_id: str) -> bool:
if directory is not None:
group = directory.get_group(group_id)
return bool(group and group.tenant_id == tenant_id and group.status == "active")
return _legacy_subject_policy(session, table_name="access_groups", tenant_id=tenant_id, scope_id=group_id) is not None
return _legacy_subject_policy(session, subject_type="group", tenant_id=tenant_id, scope_id=group_id) is not None
def _policy_scope_key(*, tenant_id: str, scope_type: str, scope_id: str | None) -> tuple[str | None, str, str | None]:
@@ -356,10 +360,10 @@ def _legacy_mail_profile_policy(
if not scope_id:
return None
if clean_scope == "user":
legacy = _legacy_subject_policy(session, table_name="access_users", tenant_id=tenant_id, scope_id=scope_id)
legacy = _legacy_subject_policy(session, subject_type="user", tenant_id=tenant_id, scope_id=scope_id)
return _policy_from_attr(legacy) if legacy is not None else None
if clean_scope == "group":
legacy = _legacy_subject_policy(session, table_name="access_groups", tenant_id=tenant_id, scope_id=scope_id)
legacy = _legacy_subject_policy(session, subject_type="group", tenant_id=tenant_id, scope_id=scope_id)
return _policy_from_attr(legacy) if legacy is not None else None
try:
context = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id)