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

Release v0.1.2

Browse Source
This commit is contained in:
Albrecht Degering
2026-06-25 19:58:20 +02:00
parent 3cd8c45c42
commit 3d82d7b32d
40 changed files with 1877 additions and 520 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -10,8 +10,8 @@ from sqlalchemy import and_, or_
from sqlalchemy.orm import Session
from govoplan_core.admin.governance import get_system_settings
from govoplan_core.core.optional import reraise_unless_missing_package
from govoplan_core.db.models import Group, Tenant, User, UserGroupMembership
from govoplan_campaign.backend.db.models import Campaign
from govoplan_mail.backend.db.models import MailServerProfile
from govoplan_mail.backend.config import ImapConfig, SmtpConfig
from govoplan_core.security.secrets import decrypt_secret, encrypt_secret
@@ -25,20 +25,47 @@ MAIL_PROFILE_POLICY_SETTINGS_KEY = "mail_profile_policy"
PROFILE_SCOPE_TYPES = {"system", "tenant", "user", "group", "campaign"}
PROFILE_PATTERN_KEYS = ("smtp_hosts", "imap_hosts", "envelope_senders", "from_headers", "recipient_domains")
PROFILE_SCOPE_ORDER = {"system": 0, "tenant": 1, "user": 2, "group": 2, "campaign": 3}
MAIL_POLICY_LIMIT_KEYS = (
"allowed_profile_ids",
"allow_user_profiles",
"allow_group_profiles",
"allow_campaign_profiles",
"smtp_credentials.inherit",
"imap_credentials.inherit",
"whitelist.smtp_hosts",
"whitelist.imap_hosts",
"whitelist.envelope_senders",
"whitelist.from_headers",
"whitelist.recipient_domains",
"blacklist.smtp_hosts",
"blacklist.imap_hosts",
"blacklist.envelope_senders",
"blacklist.from_headers",
"blacklist.recipient_domains",
)
def default_mail_policy_lower_level_limits() -> dict[str, bool]:
return {key: True for key in MAIL_POLICY_LIMIT_KEYS}
def _campaign_model():
try:
from govoplan_campaign.backend.db.models import Campaign
except ModuleNotFoundError as exc:
reraise_unless_missing_package(exc, "govoplan_campaign")
raise MailProfileError("Campaign module is not installed") from exc
return Campaign
@dataclass(slots=True)
class EffectiveCredentialPolicy:
inherit: bool = True
allow_override: bool = True
explicit_inherit: bool = False
inherit_source: str | None = None
def as_dict(self) -> dict[str, Any]:
return {
"inherit": self.inherit,
"allow_override": self.allow_override,
}
return {"inherit": self.inherit}
@dataclass(slots=True)
@@ -51,6 +78,7 @@ class EffectiveMailProfilePolicy:
imap_credentials: EffectiveCredentialPolicy = field(default_factory=EffectiveCredentialPolicy)
whitelist_groups: dict[str, list[list[str]]] = field(default_factory=lambda: {key: [] for key in PROFILE_PATTERN_KEYS})
blacklist_patterns: dict[str, list[str]] = field(default_factory=lambda: {key: [] for key in PROFILE_PATTERN_KEYS})
allow_lower_level_limits: dict[str, bool] = field(default_factory=default_mail_policy_lower_level_limits)
source_policies: list[dict[str, Any]] = field(default_factory=list)
def profile_id_allowed(self, profile_id: str) -> bool:
@@ -87,6 +115,7 @@ class EffectiveMailProfilePolicy:
"imap_credentials": self.imap_credentials.as_dict(),
"whitelist": whitelist,
"blacklist": blacklist,
"allow_lower_level_limits": dict(self.allow_lower_level_limits),
}
@@ -95,11 +124,14 @@ def slugify_profile_name(value: str) -> str:
return slug or "mail-profile"
def _transport_payload(config: SmtpConfig | ImapConfig) -> tuple[dict[str, Any], str | None, bool]:
def _transport_payload(config: SmtpConfig | ImapConfig) -> tuple[dict[str, Any], str | None, str | None, bool, bool]:
payload = config.model_dump(mode="json")
username_was_supplied = "username" in config.model_fields_set
password_was_supplied = "password" in config.model_fields_set
username = payload.pop("username", None)
password = payload.pop("password", None)
return payload, password, password_was_supplied
payload.pop("enabled", None)
return payload, username, password, username_was_supplied, password_was_supplied
def _normalize_scope_type(scope_type: str | None) -> str:
@@ -162,11 +194,20 @@ def _normalize_credential_policy(value: Any) -> dict[str, bool | None]:
inherit = data.get("inherit")
if inherit is None:
inherit = data.get("inherit_credentials")
allow_override = data.get("allow_override")
return {
"inherit": inherit if isinstance(inherit, bool) else None,
"allow_override": allow_override if isinstance(allow_override, bool) else None,
}
return {"inherit": inherit if isinstance(inherit, bool) else None}
def _normalize_allow_lower_level_limits(value: Any, *, fill_defaults: bool = False) -> dict[str, bool]:
if value is None:
return default_mail_policy_lower_level_limits() if fill_defaults else {}
if not isinstance(value, dict):
raise MailProfileError("Mail profile lower-level limit policy must be an object")
normalized = default_mail_policy_lower_level_limits() if fill_defaults else {}
for key in MAIL_POLICY_LIMIT_KEYS:
raw_value = value.get(key)
if isinstance(raw_value, bool):
normalized[key] = raw_value
return normalized
def normalize_mail_profile_policy(payload: dict[str, Any] | None) -> dict[str, Any]:
@@ -180,6 +221,7 @@ def normalize_mail_profile_policy(payload: dict[str, Any] | None) -> dict[str, A
"imap_credentials": _normalize_credential_policy(data.get("imap_credentials")),
"whitelist": whitelist,
"blacklist": blacklist,
"allow_lower_level_limits": _normalize_allow_lower_level_limits(data.get("allow_lower_level_limits")),
}
for key in ("allow_user_profiles", "allow_group_profiles", "allow_campaign_profiles"):
value = data.get(key)
@@ -211,68 +253,117 @@ def _mail_policy_source_fields(policy: dict[str, Any], *, baseline: bool = False
if isinstance(credential, dict):
if isinstance(credential.get("inherit"), bool):
fields.append(f"{key}.inherit")
if isinstance(credential.get("allow_override"), bool):
fields.append(f"{key}.allow_override")
for kind in ("whitelist", "blacklist"):
rules = policy.get(kind) or {}
if isinstance(rules, dict):
for key in PROFILE_PATTERN_KEYS:
if _clean_string_list(rules.get(key, [])):
fields.append(f"{kind}.{key}")
allow_lower = policy.get("allow_lower_level_limits") or {}
if isinstance(allow_lower, dict):
for key in MAIL_POLICY_LIMIT_KEYS:
if isinstance(allow_lower.get(key), bool):
fields.append(f"allow_lower_level_limits.{key}")
if baseline and not fields:
fields.append("defaults")
return fields
def _mail_policy_source_policy(policy: dict[str, Any], *, baseline: bool = False) -> dict[str, Any]:
normalized = normalize_mail_profile_policy(policy)
if not baseline:
return normalized
source_policy = EffectiveMailProfilePolicy().as_dict()
allowed_ids = _meaningful_allow_patterns(normalized.get("allowed_profile_ids") or [])
if allowed_ids:
source_policy["allowed_profile_ids"] = allowed_ids
for key in ("allow_user_profiles", "allow_group_profiles", "allow_campaign_profiles"):
value = normalized.get(key)
if isinstance(value, bool):
source_policy[key] = value
for key in ("smtp_credentials", "imap_credentials"):
credential = normalized.get(key) or {}
source_credential = dict(source_policy.get(key) or {})
if isinstance(credential, dict):
if isinstance(credential.get("inherit"), bool):
source_credential["inherit"] = credential["inherit"]
source_policy[key] = source_credential
source_policy["whitelist"] = dict(normalized.get("whitelist") or {})
source_policy["blacklist"] = dict(normalized.get("blacklist") or {})
source_policy["allow_lower_level_limits"] = {
**default_mail_policy_lower_level_limits(),
**(normalized.get("allow_lower_level_limits") or {}),
}
return source_policy
def _mail_policy_source_step(source: str, source_id: str | None, label: str, policy: dict[str, Any], *, baseline: bool = False) -> dict[str, Any]:
return {
"scope_type": source,
"scope_id": source_id,
"label": label,
"applied_fields": _mail_policy_source_fields(policy, baseline=baseline),
"policy": _mail_policy_source_policy(policy, baseline=baseline),
}
def _merge_credential_policy(current: EffectiveCredentialPolicy, data: dict[str, Any], *, source: str) -> None:
inherit = data.get("inherit")
if isinstance(inherit, bool):
if current.allow_override:
current.inherit = inherit
current.explicit_inherit = True
current.inherit_source = source
elif inherit == current.inherit:
current.explicit_inherit = True
allow_override = data.get("allow_override")
if isinstance(allow_override, bool):
if current.allow_override:
current.allow_override = allow_override
elif allow_override is False:
current.allow_override = False
def _merge_credential_policy(
current: EffectiveCredentialPolicy,
data: dict[str, Any],
*,
source: str,
lower_limits: dict[str, bool],
inherit_limit_key: str,
) -> None:
inherit = data.get("inherit")
if isinstance(inherit, bool) and lower_limits.get(inherit_limit_key, True):
current.inherit = inherit
current.explicit_inherit = True
current.inherit_source = source
def _merge_policy(policy: EffectiveMailProfilePolicy, data: dict[str, Any], *, source: str, source_id: str | None = None, label: str | None = None, baseline: bool = False) -> None:
normalized = normalize_mail_profile_policy(data)
policy.source_policies.append(_mail_policy_source_step(source, source_id, label or source.capitalize(), normalized, baseline=baseline))
lower_limits = dict(policy.allow_lower_level_limits)
allowed_ids = _meaningful_allow_patterns(normalized.get("allowed_profile_ids") or [])
if allowed_ids:
if allowed_ids and lower_limits.get("allowed_profile_ids", True):
policy.allowed_profile_id_sets.append(set(allowed_ids))
if normalized.get("allow_user_profiles") is False:
if normalized.get("allow_user_profiles") is False and lower_limits.get("allow_user_profiles", True):
policy.allow_user_profiles = False
if normalized.get("allow_group_profiles") is False:
if normalized.get("allow_group_profiles") is False and lower_limits.get("allow_group_profiles", True):
policy.allow_group_profiles = False
if normalized.get("allow_campaign_profiles") is False:
if normalized.get("allow_campaign_profiles") is False and lower_limits.get("allow_campaign_profiles", True):
policy.allow_campaign_profiles = False
_merge_credential_policy(policy.smtp_credentials, normalized.get("smtp_credentials") or {}, source=source)
_merge_credential_policy(policy.imap_credentials, normalized.get("imap_credentials") or {}, source=source)
_merge_credential_policy(
policy.smtp_credentials,
normalized.get("smtp_credentials") or {},
source=source,
lower_limits=lower_limits,
inherit_limit_key="smtp_credentials.inherit",
)
_merge_credential_policy(
policy.imap_credentials,
normalized.get("imap_credentials") or {},
source=source,
lower_limits=lower_limits,
inherit_limit_key="imap_credentials.inherit",
)
whitelist = normalized.get("whitelist") or {}
blacklist = normalized.get("blacklist") or {}
for key in PROFILE_PATTERN_KEYS:
allow_patterns = _meaningful_allow_patterns(whitelist.get(key, []))
if allow_patterns:
if allow_patterns and lower_limits.get(f"whitelist.{key}", True):
policy.whitelist_groups.setdefault(key, []).append(allow_patterns)
deny_patterns = _clean_string_list(blacklist.get(key, []))
if deny_patterns:
if deny_patterns and lower_limits.get(f"blacklist.{key}", True):
policy.blacklist_patterns.setdefault(key, []).extend(deny_patterns)
local_lower_limits = normalized.get("allow_lower_level_limits") or {}
if local_lower_limits:
policy.allow_lower_level_limits = {
key: lower_limits.get(key, True) and local_lower_limits.get(key, lower_limits.get(key, True))
for key in MAIL_POLICY_LIMIT_KEYS
}
def _owner_context(
@@ -282,10 +373,10 @@ def _owner_context(
campaign_id: str | None = None,
owner_user_id: str | None = None,
owner_group_id: str | None = None,
) -> tuple[Campaign | None, str | None, str | None]:
) -> tuple[Any | None, str | None, str | None]:
campaign = None
if campaign_id:
campaign = session.get(Campaign, campaign_id)
campaign = session.get(_campaign_model(), campaign_id)
if campaign is None or campaign.tenant_id != tenant_id:
raise MailProfileError("Campaign not found for mail-server profile policy")
owner_user_id = campaign.owner_user_id
@@ -329,6 +420,38 @@ def effective_mail_profile_policy(
return policy
def effective_mail_profile_policy_for_scope(
session: Session,
*,
tenant_id: str,
scope_type: str,
scope_id: str | None = None,
campaign_id: str | None = None,
) -> EffectiveMailProfilePolicy:
clean_scope = _normalize_scope_type(scope_type)
if campaign_id:
return effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=campaign_id)
if clean_scope == "system":
policy = EffectiveMailProfilePolicy()
system_settings = get_system_settings(session)
_merge_policy(policy, _policy_from_settings(system_settings.settings or {}), source="system", label="System", baseline=True)
return policy
if clean_scope == "tenant":
return effective_mail_profile_policy(session, tenant_id=tenant_id)
if clean_scope == "user":
if not scope_id:
raise MailProfileError("User mail profile policy requires scope_id")
return effective_mail_profile_policy(session, tenant_id=tenant_id, owner_user_id=scope_id)
if clean_scope == "group":
if not scope_id:
raise MailProfileError("Group mail profile policy requires scope_id")
return effective_mail_profile_policy(session, tenant_id=tenant_id, owner_group_id=scope_id)
if not scope_id:
raise MailProfileError("Campaign mail profile policy requires scope_id")
return effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=scope_id)
def parent_mail_profile_policy(
session: Session,
*,
@@ -353,7 +476,7 @@ def parent_mail_profile_policy(
if clean_scope != "campaign" or not scope_id:
return policy
campaign = session.get(Campaign, scope_id)
campaign = session.get(_campaign_model(), scope_id)
if campaign is None or campaign.tenant_id != tenant_id:
raise MailProfileError("Campaign policy not found")
if campaign.owner_user_id:
@@ -470,7 +593,7 @@ def _scope_tuple_for_create(
return tenant_id, clean_scope, clean_scope_id
if not clean_scope_id:
raise MailProfileError("Campaign-scoped mail-server profiles require a campaign scope_id")
campaign = session.get(Campaign, clean_scope_id)
campaign = session.get(_campaign_model(), clean_scope_id)
if campaign is None or campaign.tenant_id != tenant_id:
raise MailProfileError("Campaign scope is not part of the active tenant")
return tenant_id, clean_scope, clean_scope_id
@@ -524,7 +647,7 @@ def _profile_visible_filter(tenant_id: str):
return or_(MailServerProfile.scope_type == "system", MailServerProfile.tenant_id == tenant_id)
def _context_profile_clauses(campaign: Campaign, policy: EffectiveMailProfilePolicy) -> list[Any]:
def _context_profile_clauses(campaign: Any, policy: EffectiveMailProfilePolicy) -> list[Any]:
clauses: list[Any] = [
MailServerProfile.scope_type == "system",
and_(MailServerProfile.tenant_id == campaign.tenant_id, MailServerProfile.scope_type == "tenant"),
@@ -551,7 +674,7 @@ def _profile_allowed_for_context(
*,
tenant_id: str,
policy: EffectiveMailProfilePolicy,
campaign: Campaign | None = None,
campaign: Any | None = None,
owner_user_id: str | None = None,
owner_group_id: str | None = None,
) -> bool:
@@ -588,7 +711,7 @@ def list_mail_server_profiles(
campaign_id: str | None = None,
) -> list[MailServerProfile]:
if campaign_id:
campaign = session.get(Campaign, campaign_id)
campaign = session.get(_campaign_model(), campaign_id)
if campaign is None or campaign.tenant_id != tenant_id:
raise MailProfileError("Campaign not found for mail-server profiles")
policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=campaign.id)
@@ -639,7 +762,7 @@ def ensure_mail_profile_allowed_for_campaign(
profile_id: str,
require_active: bool = True,
) -> MailServerProfile:
campaign = session.get(Campaign, campaign_id)
campaign = session.get(_campaign_model(), campaign_id)
if campaign is None or campaign.tenant_id != tenant_id:
raise MailProfileError("Campaign not found for mail-server profile policy")
profile = get_mail_server_profile(session, tenant_id=tenant_id, profile_id=profile_id, require_active=require_active)
@@ -668,10 +791,18 @@ def _legacy_profile_credentials_inherited(server: dict[str, Any], protocol: str)
return server.get(f"inherit_{protocol}_credentials") is not False
def _local_credential_inheritance_override_allowed(policy: EffectiveMailProfilePolicy, protocol: str) -> bool:
credential_policy = _credential_policy_for_protocol(policy, protocol)
return (
credential_policy.inherit_source != "campaign"
and policy.allow_lower_level_limits.get(f"{protocol}_credentials.inherit", True)
)
def profile_credentials_inherited_for_server(policy: EffectiveMailProfilePolicy, server: dict[str, Any], protocol: str) -> bool:
credential_policy = _credential_policy_for_protocol(policy, protocol)
legacy_key = f"inherit_{protocol}_credentials"
if legacy_key in server and credential_policy.allow_override and credential_policy.inherit_source != "campaign":
if legacy_key in server and _local_credential_inheritance_override_allowed(policy, protocol):
return _legacy_profile_credentials_inherited(server, protocol)
return credential_policy.inherit
@@ -718,7 +849,7 @@ def _assert_profile_credentials_allowed_for_server(profile: MailServerProfile, p
legacy_key = f"inherit_{protocol}_credentials"
if legacy_key in server:
legacy_inherit = _legacy_profile_credentials_inherited(server, protocol)
if (not credential_policy.allow_override or credential_policy.inherit_source == "campaign") and legacy_inherit != credential_policy.inherit:
if not _local_credential_inheritance_override_allowed(policy, protocol) and legacy_inherit != credential_policy.inherit:
raise MailProfileError(f"{protocol.upper()} credential inheritance is locked by the effective mail profile policy")
inherited = profile_credentials_inherited_for_server(policy, server, protocol)
username, password = _transport_credentials_from_server(server, protocol)
@@ -728,14 +859,16 @@ def _assert_profile_credentials_allowed_for_server(profile: MailServerProfile, p
raise MailProfileError(f"{protocol.upper()} credentials must be supplied by this campaign or scope policy")
def _inline_transport_configured(value: Any) -> bool:
if not isinstance(value, dict):
return False
if value.get("enabled") is True:
return True
for key in ("host", "username", "password"):
def _inline_transport_configured(server: dict[str, Any], protocol: str) -> bool:
value = server.get(protocol) if isinstance(server.get(protocol), dict) else {}
credentials = server.get("credentials") if isinstance(server.get("credentials"), dict) else {}
credential_value = credentials.get(protocol) if isinstance(credentials.get(protocol), dict) else {}
for key in ("host",):
if str(value.get(key) or "").strip():
return True
for key in ("username", "password"):
if str(credential_value.get(key) or value.get(key) or "").strip():
return True
return False
@@ -784,7 +917,7 @@ def assert_campaign_mail_policy_allows_json(
imap = server.get("imap") if isinstance(server, dict) and isinstance(server.get("imap"), dict) else None
if campaign_id:
policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=campaign_id)
if not policy.allow_campaign_profiles and (_inline_transport_configured(smtp) or _inline_transport_configured(imap)):
if not policy.allow_campaign_profiles and (_inline_transport_configured(server, "smtp") or _inline_transport_configured(server, "imap")):
raise MailProfileError("Campaign-local inline mail settings are disabled by policy")
_assert_transport_values_allowed(policy, smtp, imap)
return
@@ -834,11 +967,12 @@ def create_mail_server_profile(
assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=smtp, imap=imap)
clean_slug = slugify_profile_name(slug or clean_name)
_ensure_unique_slug(session, scope_type=clean_scope_type, scope_id=clean_scope_id, slug=clean_slug)
smtp_payload, smtp_password, _ = _transport_payload(smtp)
smtp_payload, smtp_username, smtp_password, _, _ = _transport_payload(smtp)
imap_payload = None
imap_username = None
imap_password = None
if imap is not None:
imap_payload, imap_password, _ = _transport_payload(imap)
imap_payload, imap_username, imap_password, _, _ = _transport_payload(imap)
profile = MailServerProfile(
tenant_id=profile_tenant_id,
scope_type=clean_scope_type,
@@ -848,8 +982,10 @@ def create_mail_server_profile(
description=description,
is_active=is_active,
smtp_config=smtp_payload,
smtp_username=smtp_username,
smtp_password_encrypted=encrypt_secret(smtp_password),
imap_config=imap_payload,
imap_username=imap_username,
imap_password_encrypted=encrypt_secret(imap_password),
created_by_user_id=user_id,
updated_by_user_id=user_id,
@@ -893,13 +1029,13 @@ def update_mail_server_profile(
if is_active is not None:
profile.is_active = is_active
next_smtp = smtp or SmtpConfig.model_validate({**(profile.smtp_config or {}), "password": decrypt_secret(profile.smtp_password_encrypted)})
next_smtp = smtp or SmtpConfig.model_validate({**(profile.smtp_config or {}), "username": _profile_username(profile, "smtp"), "password": decrypt_secret(profile.smtp_password_encrypted)})
if clear_imap:
next_imap = None
elif imap is not None:
next_imap = imap
elif profile.imap_config:
next_imap = ImapConfig.model_validate({**(profile.imap_config or {}), "password": decrypt_secret(profile.imap_password_encrypted)})
next_imap = ImapConfig.model_validate({**(profile.imap_config or {}), "username": _profile_username(profile, "imap"), "password": decrypt_secret(profile.imap_password_encrypted)})
else:
next_imap = None
@@ -915,17 +1051,22 @@ def update_mail_server_profile(
assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=next_smtp, imap=next_imap)
if smtp is not None:
smtp_payload, smtp_password, supplied = _transport_payload(smtp)
smtp_payload, smtp_username, smtp_password, username_supplied, password_supplied = _transport_payload(smtp)
profile.smtp_config = smtp_payload
if supplied:
if username_supplied:
profile.smtp_username = smtp_username
if password_supplied:
profile.smtp_password_encrypted = encrypt_secret(smtp_password)
if clear_imap:
profile.imap_config = None
profile.imap_username = None
profile.imap_password_encrypted = None
elif imap is not None:
imap_payload, imap_password, supplied = _transport_payload(imap)
imap_payload, imap_username, imap_password, username_supplied, password_supplied = _transport_payload(imap)
profile.imap_config = imap_payload
if supplied:
if username_supplied:
profile.imap_username = imap_username
if password_supplied:
profile.imap_password_encrypted = encrypt_secret(imap_password)
profile.updated_by_user_id = user_id
session.add(profile)
@@ -933,8 +1074,18 @@ def update_mail_server_profile(
return profile
def _profile_username(profile: MailServerProfile, protocol: str) -> str | None:
if protocol == "smtp":
return profile.smtp_username or (profile.smtp_config or {}).get("username")
if protocol == "imap":
return profile.imap_username or (profile.imap_config or {}).get("username")
return None
def smtp_config_from_profile(profile: MailServerProfile) -> SmtpConfig:
payload = dict(profile.smtp_config or {})
payload.pop("username", None)
payload["username"] = _profile_username(profile, "smtp")
payload["password"] = decrypt_secret(profile.smtp_password_encrypted)
return SmtpConfig.model_validate(payload)
@@ -943,6 +1094,9 @@ def imap_config_from_profile(profile: MailServerProfile) -> ImapConfig | None:
if not profile.imap_config:
return None
payload = dict(profile.imap_config or {})
payload.pop("username", None)
payload.pop("enabled", None)
payload["username"] = _profile_username(profile, "imap")
payload["password"] = decrypt_secret(profile.imap_password_encrypted)
return ImapConfig.model_validate(payload)
@@ -952,11 +1106,15 @@ def inherit_profile_credentials(server: dict[str, Any], protocol: str) -> bool:
def _transport_credentials_from_server(server: dict[str, Any], protocol: str) -> tuple[str | None, str | None]:
config = server.get(protocol)
if not isinstance(config, dict):
return None, None
username = config.get("username")
password = config.get("password")
credentials = server.get("credentials") if isinstance(server.get("credentials"), dict) else {}
config = credentials.get(protocol) if isinstance(credentials.get(protocol), dict) else None
legacy_config = server.get(protocol) if isinstance(server.get(protocol), dict) else None
username = config.get("username") if config is not None else None
password = config.get("password") if config is not None else None
if username in (None, "") and legacy_config is not None:
username = legacy_config.get("username")
if password in (None, "") and legacy_config is not None:
password = legacy_config.get("password")
return (str(username) if username not in (None, "") else None, str(password) if password not in (None, "") else None)
@@ -967,6 +1125,14 @@ def apply_campaign_credentials(payload: dict[str, Any], server: dict[str, Any],
return payload
def _server_config_payload(value: dict[str, Any] | None) -> dict[str, Any]:
payload = dict(value or {})
payload.pop("username", None)
payload.pop("password", None)
payload.pop("enabled", None)
return payload
def profile_response_payload(profile: MailServerProfile) -> dict[str, Any]:
return {
"id": profile.id,
@@ -977,8 +1143,12 @@ def profile_response_payload(profile: MailServerProfile) -> dict[str, Any]:
"slug": profile.slug,
"description": profile.description,
"is_active": profile.is_active,
"smtp": dict(profile.smtp_config or {}),
"imap": dict(profile.imap_config or {}) if profile.imap_config else None,
"smtp": _server_config_payload(profile.smtp_config),
"imap": _server_config_payload(profile.imap_config) if profile.imap_config else None,
"credentials": {
"smtp": {"username": _profile_username(profile, "smtp")},
"imap": {"username": _profile_username(profile, "imap")},
},
"smtp_password_configured": bool(profile.smtp_password_encrypted),
"imap_password_configured": bool(profile.imap_password_encrypted),
"created_at": profile.created_at,
@@ -1013,22 +1183,36 @@ def get_mail_profile_policy(
if group is None or group.tenant_id != tenant_id:
raise MailProfileError("Group policy not found")
return _policy_from_attr(group.mail_profile_policy)
campaign = session.get(Campaign, scope_id)
campaign = session.get(_campaign_model(), scope_id)
if campaign is None or campaign.tenant_id != tenant_id:
raise MailProfileError("Campaign policy not found")
return _policy_from_attr(campaign.mail_profile_policy)
def _validate_policy_against_parent(parent: EffectiveMailProfilePolicy, normalized: dict[str, Any]) -> None:
parent_limits = parent.allow_lower_level_limits
allowed_ids = _meaningful_allow_patterns(normalized.get("allowed_profile_ids") or [])
if allowed_ids and not parent_limits.get("allowed_profile_ids", True):
raise MailProfileError("Mail profile allow-list is locked by an ancestor policy")
for key in ("allow_user_profiles", "allow_group_profiles", "allow_campaign_profiles"):
if isinstance(normalized.get(key), bool) and not parent_limits.get(key, True):
raise MailProfileError(f"{key} is locked by an ancestor policy")
for kind in ("whitelist", "blacklist"):
rules = normalized.get(kind) or {}
if isinstance(rules, dict):
for key in PROFILE_PATTERN_KEYS:
if _clean_string_list(rules.get(key, [])) and not parent_limits.get(f"{kind}.{key}", True):
raise MailProfileError(f"{kind}.{key} is locked by an ancestor policy")
for protocol in ("smtp", "imap"):
local = normalized.get(f"{protocol}_credentials") or {}
parent_credentials = _credential_policy_for_protocol(parent, protocol)
local_inherit = local.get("inherit")
if isinstance(local_inherit, bool) and not parent_credentials.allow_override and local_inherit != parent_credentials.inherit:
if isinstance(local_inherit, bool) and not parent_limits.get(f"{protocol}_credentials.inherit", True):
raise MailProfileError(f"{protocol.upper()} credential inheritance is locked by an ancestor policy")
local_allow_override = local.get("allow_override")
if local_allow_override is True and not parent_credentials.allow_override:
raise MailProfileError(f"{protocol.upper()} credential override cannot be re-enabled below a locked ancestor policy")
local_lower_limits = normalized.get("allow_lower_level_limits") or {}
if isinstance(local_lower_limits, dict):
for key, allowed in local_lower_limits.items():
if allowed is True and not parent_limits.get(key, True):
raise MailProfileError(f"{key} lower-level limiting cannot be re-enabled below a locked ancestor policy")
def set_mail_profile_policy(
@@ -1078,7 +1262,7 @@ def set_mail_profile_policy(
group.mail_profile_policy = normalized
session.add(group)
return normalized
campaign = session.get(Campaign, scope_id)
campaign = session.get(_campaign_model(), scope_id)
if campaign is None or campaign.tenant_id != tenant_id:
raise MailProfileError("Campaign policy not found")
campaign.mail_profile_policy = normalized
@@ -1086,6 +1270,23 @@ def set_mail_profile_policy(
return normalized
def _write_transport_to_server(server: dict[str, Any], protocol: str, payload: dict[str, Any]) -> None:
server_payload = dict(payload)
username = server_payload.pop("username", None)
password = server_payload.pop("password", None)
server_payload.pop("enabled", None)
server[protocol] = server_payload
if username not in (None, "") or password not in (None, ""):
credentials = server.get("credentials") if isinstance(server.get("credentials"), dict) else {}
protocol_credentials = credentials.get(protocol) if isinstance(credentials.get(protocol), dict) else {}
if username not in (None, ""):
protocol_credentials["username"] = username
if password not in (None, ""):
protocol_credentials["password"] = password
credentials[protocol] = protocol_credentials
server["credentials"] = credentials
def materialize_campaign_mail_profile_config(
session: Session,
*,
@@ -1133,13 +1334,13 @@ def materialize_campaign_mail_profile_config(
smtp_payload = smtp_config_from_profile(profile).model_dump(mode="json")
if not profile_credentials_inherited_for_server(policy, server, "smtp"):
smtp_payload = apply_campaign_credentials(smtp_payload, server, "smtp")
resolved_server["smtp"] = smtp_payload
_write_transport_to_server(resolved_server, "smtp", smtp_payload)
imap = imap_config_from_profile(profile)
if imap is not None:
imap_payload = imap.model_dump(mode="json")
if not profile_credentials_inherited_for_server(policy, server, "imap"):
imap_payload = apply_campaign_credentials(imap_payload, server, "imap")
resolved_server["imap"] = imap_payload
_write_transport_to_server(resolved_server, "imap", imap_payload)
else:
resolved_server.pop("imap", None)
data["server"] = resolved_server