from __future__ import annotations import fnmatch import json import re from dataclasses import dataclass, field from typing import Any, Iterable from sqlalchemy import and_, or_, select, text from sqlalchemy.orm import Session from govoplan_core.admin.settings import get_system_settings from govoplan_core.audit.logging import audit_event from govoplan_core.core.access import CAPABILITY_ACCESS_DIRECTORY, AccessDirectory from govoplan_core.core.campaigns import ( CAPABILITY_CAMPAIGNS_ACCESS, CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT, CampaignAccessProvider, CampaignMailPolicyContext, CampaignMailPolicyContextProvider, ) from govoplan_core.core.policy import policy_source_step from govoplan_core.core.runtime import get_registry from govoplan_mail.backend.db.models import MailProfilePolicy, MailServerProfile, new_uuid from govoplan_mail.backend.config import ImapConfig, SmtpConfig from govoplan_mail.backend.mailbox_index import clear_mailbox_index from govoplan_core.security.secrets import decrypt_secret, encrypt_secret class MailProfileError(RuntimeError): pass 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} @dataclass(slots=True) class EffectiveCredentialPolicy: inherit: bool = True explicit_inherit: bool = False inherit_source: str | None = None def as_dict(self) -> dict[str, Any]: return {"inherit": self.inherit} @dataclass(slots=True) class EffectiveMailProfilePolicy: allowed_profile_id_sets: list[set[str]] = field(default_factory=list) allow_user_profiles: bool = True allow_group_profiles: bool = True allow_campaign_profiles: bool = True smtp_credentials: EffectiveCredentialPolicy = field(default_factory=EffectiveCredentialPolicy) 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: return all(profile_id in allowed for allowed in self.allowed_profile_id_sets) def value_allowed(self, key: str, value: str | None) -> tuple[bool, str | None]: if key not in PROFILE_PATTERN_KEYS or not value: return True, None normalized = _normalize_match_value(value) for pattern in self.blacklist_patterns.get(key, []): if _pattern_matches(normalized, pattern): return False, f"{key} value {value!r} is blocked by mail profile blacklist pattern {pattern!r}" for group in self.whitelist_groups.get(key, []): if group and not any(_pattern_matches(normalized, pattern) for pattern in group): return False, f"{key} value {value!r} is not allowed by the effective mail profile whitelist" return True, None def as_dict(self) -> dict[str, Any]: allowed_profile_ids: list[str] | None = None if self.allowed_profile_id_sets: allowed_profile_ids = sorted(set.intersection(*self.allowed_profile_id_sets)) if len(self.allowed_profile_id_sets) > 1 else sorted(next(iter(self.allowed_profile_id_sets))) whitelist: dict[str, list[str]] = {} for key, groups in self.whitelist_groups.items(): flattened = sorted({pattern for group in groups for pattern in group}) if flattened: whitelist[key] = flattened blacklist = {key: patterns for key, patterns in self.blacklist_patterns.items() if patterns} return { "allowed_profile_ids": allowed_profile_ids, "allow_user_profiles": self.allow_user_profiles, "allow_group_profiles": self.allow_group_profiles, "allow_campaign_profiles": self.allow_campaign_profiles, "smtp_credentials": self.smtp_credentials.as_dict(), "imap_credentials": self.imap_credentials.as_dict(), "whitelist": whitelist, "blacklist": blacklist, "allow_lower_level_limits": dict(self.allow_lower_level_limits), } def slugify_profile_name(value: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", value.strip().casefold()).strip("-") return slug or "mail-profile" 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) payload.pop("enabled", None) return payload, username, password, username_was_supplied, password_was_supplied def _normalize_scope_type(scope_type: str | None) -> str: clean = (scope_type or "tenant").strip().casefold() if clean not in PROFILE_SCOPE_TYPES: raise MailProfileError("Mail-server profile scope must be system, tenant, user, group or campaign") return clean def _profile_scope_type(profile: MailServerProfile) -> str: return _normalize_scope_type(getattr(profile, "scope_type", None) or "tenant") def _profile_scope_id(profile: MailServerProfile) -> str | None: scope_type = _profile_scope_type(profile) value = getattr(profile, "scope_id", None) if value: return str(value) if scope_type == "tenant" and profile.tenant_id: return str(profile.tenant_id) return None def _clean_string_list(value: Any) -> list[str]: if value is None: return [] if isinstance(value, str): value = [value] if not isinstance(value, list): raise MailProfileError("Mail profile policy lists must be arrays of strings") result: list[str] = [] for item in value: text = str(item).strip() if text: result.append(text) return result def _meaningful_allow_patterns(patterns: Iterable[str]) -> list[str]: return [pattern for pattern in patterns if pattern and pattern != "*"] def _normalize_pattern_rules(value: Any) -> dict[str, list[str]]: if value is None: return {} if not isinstance(value, dict): raise MailProfileError("Mail profile pattern policy must be an object") rules: dict[str, list[str]] = {} for key, raw_patterns in value.items(): if key not in PROFILE_PATTERN_KEYS: raise MailProfileError(f"Unsupported mail profile pattern key: {key}") patterns = _clean_string_list(raw_patterns) if patterns: rules[key] = patterns return rules def _normalize_credential_policy(value: Any) -> dict[str, bool | None]: data = value if isinstance(value, dict) else {} inherit = data.get("inherit") if inherit is None: inherit = data.get("inherit_credentials") 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]: data = dict(payload or {}) allowed_ids = _clean_string_list(data.get("allowed_profile_ids")) whitelist = _normalize_pattern_rules(data.get("whitelist") or data.get("allow_patterns") or {}) blacklist = _normalize_pattern_rules(data.get("blacklist") or data.get("deny_patterns") or {}) normalized: dict[str, Any] = { "allowed_profile_ids": allowed_ids, "smtp_credentials": _normalize_credential_policy(data.get("smtp_credentials")), "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) normalized[key] = value if isinstance(value, bool) else None return normalized def _policy_from_settings(settings_payload: dict[str, Any] | None) -> dict[str, Any]: if not isinstance(settings_payload, dict): return normalize_mail_profile_policy({}) raw = settings_payload.get(MAIL_PROFILE_POLICY_SETTINGS_KEY, {}) return normalize_mail_profile_policy(raw if isinstance(raw, dict) else {}) def _policy_from_attr(value: Any) -> dict[str, Any]: return normalize_mail_profile_policy(value if isinstance(value, dict) else {}) def _json_object(value: Any) -> dict[str, Any]: if isinstance(value, dict): return value if isinstance(value, str) and value.strip(): try: parsed = json.loads(value) except json.JSONDecodeError: return {} return parsed if isinstance(parsed, dict) else {} return {} def _access_directory() -> AccessDirectory | None: registry = get_registry() if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_ACCESS_DIRECTORY): return None capability = registry.require_capability(CAPABILITY_ACCESS_DIRECTORY) if not isinstance(capability, AccessDirectory): raise MailProfileError("Access directory capability is invalid") return capability def _campaign_policy_provider() -> CampaignMailPolicyContextProvider | None: registry = get_registry() if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT): return None capability = registry.require_capability(CAPABILITY_CAMPAIGNS_MAIL_POLICY_CONTEXT) if not isinstance(capability, CampaignMailPolicyContextProvider): raise MailProfileError("Campaign mail policy context capability is invalid") return capability def _campaign_access_provider() -> CampaignAccessProvider | None: registry = get_registry() if registry is None or not hasattr(registry, "has_capability") or not registry.has_capability(CAPABILITY_CAMPAIGNS_ACCESS): return None capability = registry.require_capability(CAPABILITY_CAMPAIGNS_ACCESS) if not isinstance(capability, CampaignAccessProvider): raise MailProfileError("Campaign access capability is invalid") return capability def _campaign_policy_context(session: Session, *, tenant_id: str, campaign_id: str) -> CampaignMailPolicyContext: provider = _campaign_policy_provider() if provider is None: raise MailProfileError("Campaign module is not available") context = provider.get_campaign_mail_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_id) if context is None: raise MailProfileError("Campaign not found for mail-server profile policy") return context def _legacy_tenant_settings(session: Session, tenant_id: str) -> dict[str, Any] | None: row = session.execute(text("select settings from core_scopes where id = :tenant_id"), {"tenant_id": tenant_id}).first() return _json_object(row[0]) if row else None 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( statement, {"scope_id": scope_id, "tenant_id": tenant_id}, ).first() return _json_object(row[0]) if row else None def _tenant_exists(session: Session, tenant_id: str) -> bool: return _legacy_tenant_settings(session, tenant_id) is not None def _user_exists(session: Session, *, tenant_id: str, user_id: str) -> bool: directory = _access_directory() 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, 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: directory = _access_directory() 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, 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]: clean_scope = _normalize_scope_type(scope_type) if clean_scope == "system": return None, clean_scope, None if clean_scope == "tenant": return tenant_id, clean_scope, scope_id or tenant_id return tenant_id, clean_scope, scope_id def _mail_policy_row( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None, ) -> MailProfilePolicy | None: row_tenant_id, row_scope_type, row_scope_id = _policy_scope_key(tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id) query = session.query(MailProfilePolicy).filter(MailProfilePolicy.scope_type == row_scope_type) query = query.filter(MailProfilePolicy.tenant_id.is_(None) if row_tenant_id is None else MailProfilePolicy.tenant_id == row_tenant_id) query = query.filter(MailProfilePolicy.scope_id.is_(None) if row_scope_id is None else MailProfilePolicy.scope_id == row_scope_id) return query.one_or_none() def _legacy_mail_profile_policy( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None, ) -> dict[str, Any] | None: clean_scope = _normalize_scope_type(scope_type) if clean_scope == "system": return _policy_from_settings(get_system_settings(session).settings or {}) if clean_scope == "tenant": if scope_id and scope_id != tenant_id: return None settings = _legacy_tenant_settings(session, tenant_id) return _policy_from_settings(settings or {}) if settings is not None else None if not scope_id: return None if clean_scope == "user": 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, 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) except MailProfileError: return None return _policy_from_attr(context.mail_profile_policy) def _read_mail_profile_policy( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None, ) -> dict[str, Any] | None: clean_scope = _normalize_scope_type(scope_type) if clean_scope != "campaign": row = _mail_policy_row(session, tenant_id=tenant_id, scope_type=clean_scope, scope_id=scope_id) if row is not None: return _policy_from_attr(row.policy) return _legacy_mail_profile_policy(session, tenant_id=tenant_id, scope_type=clean_scope, scope_id=scope_id) def _write_mail_profile_policy( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None, policy: dict[str, Any], ) -> dict[str, Any]: row_tenant_id, row_scope_type, row_scope_id = _policy_scope_key(tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id) row = _mail_policy_row(session, tenant_id=tenant_id, scope_type=row_scope_type, scope_id=row_scope_id) if row is None: row = MailProfilePolicy( tenant_id=row_tenant_id, scope_type=row_scope_type, scope_id=row_scope_id, policy=policy, ) else: row.policy = policy session.add(row) return policy def _mail_policy_source_fields(policy: dict[str, Any], *, baseline: bool = False) -> list[str]: fields: list[str] = [] if _meaningful_allow_patterns(policy.get("allowed_profile_ids") or []): fields.append("allowed_profile_ids") for key in ("allow_user_profiles", "allow_group_profiles", "allow_campaign_profiles"): if isinstance(policy.get(key), bool): fields.append(key) for key in ("smtp_credentials", "imap_credentials"): credential = policy.get(key) or {} if isinstance(credential, dict): if isinstance(credential.get("inherit"), bool): fields.append(f"{key}.inherit") 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 policy_source_step( source, label, source_id, 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, 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) _merge_profile_definition_policy(policy, normalized, lower_limits) _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", ) _merge_pattern_policy(policy, normalized, lower_limits) _merge_lower_level_limits(policy, normalized, lower_limits) def _policy_limit_enabled(lower_limits: dict[str, bool], key: str) -> bool: return lower_limits.get(key, True) def _merge_profile_definition_policy(policy: EffectiveMailProfilePolicy, normalized: dict[str, Any], lower_limits: dict[str, bool]) -> None: allowed_ids = _meaningful_allow_patterns(normalized.get("allowed_profile_ids") or []) if allowed_ids and _policy_limit_enabled(lower_limits, "allowed_profile_ids"): policy.allowed_profile_id_sets.append(set(allowed_ids)) if normalized.get("allow_user_profiles") is False and _policy_limit_enabled(lower_limits, "allow_user_profiles"): policy.allow_user_profiles = False if normalized.get("allow_group_profiles") is False and _policy_limit_enabled(lower_limits, "allow_group_profiles"): policy.allow_group_profiles = False if normalized.get("allow_campaign_profiles") is False and _policy_limit_enabled(lower_limits, "allow_campaign_profiles"): policy.allow_campaign_profiles = False def _merge_pattern_policy(policy: EffectiveMailProfilePolicy, normalized: dict[str, Any], lower_limits: dict[str, bool]) -> None: 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 and _policy_limit_enabled(lower_limits, f"whitelist.{key}"): policy.whitelist_groups.setdefault(key, []).append(allow_patterns) deny_patterns = _clean_string_list(blacklist.get(key, [])) if deny_patterns and _policy_limit_enabled(lower_limits, f"blacklist.{key}"): policy.blacklist_patterns.setdefault(key, []).extend(deny_patterns) def _merge_lower_level_limits(policy: EffectiveMailProfilePolicy, normalized: dict[str, Any], lower_limits: dict[str, bool]) -> None: local_lower_limits = normalized.get("allow_lower_level_limits") or {} if local_lower_limits: policy.allow_lower_level_limits = { key: _policy_limit_enabled(lower_limits, key) and local_lower_limits.get(key, _policy_limit_enabled(lower_limits, key)) for key in MAIL_POLICY_LIMIT_KEYS } def _owner_context( session: Session, *, tenant_id: str, campaign_id: str | None = None, owner_user_id: str | None = None, owner_group_id: str | None = None, ) -> tuple[CampaignMailPolicyContext | None, str | None, str | None]: campaign_context = None if campaign_id: campaign_context = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_id) owner_user_id = campaign_context.owner_user_id owner_group_id = campaign_context.owner_group_id return campaign_context, owner_user_id, owner_group_id def effective_mail_profile_policy( session: Session, *, tenant_id: str, campaign_id: str | None = None, owner_user_id: str | None = None, owner_group_id: str | None = None, ) -> EffectiveMailProfilePolicy: campaign, owner_user_id, owner_group_id = _owner_context( session, tenant_id=tenant_id, campaign_id=campaign_id, owner_user_id=owner_user_id, owner_group_id=owner_group_id, ) tenant_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="tenant", scope_id=tenant_id) if tenant_policy is None: raise MailProfileError("Tenant not found for mail-server profile policy") policy = EffectiveMailProfilePolicy() system_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="system") or {} _merge_policy(policy, system_policy, source="system", label="System", baseline=True) _merge_policy(policy, tenant_policy, source="tenant", source_id=tenant_id, label="Tenant") if owner_user_id: user_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="user", scope_id=owner_user_id) if user_policy is not None: _merge_policy(policy, user_policy, source="user", source_id=owner_user_id, label="Owner user") if owner_group_id: group_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="group", scope_id=owner_group_id) if group_policy is not None: _merge_policy(policy, group_policy, source="group", source_id=owner_group_id, label="Owner group") if campaign is not None: _merge_policy(policy, _policy_from_attr(campaign.mail_profile_policy), source="campaign", source_id=campaign.id, label="Campaign") 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_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="system") or {} _merge_policy(policy, system_policy, 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, *, tenant_id: str, scope_type: str, scope_id: str | None = None, ) -> EffectiveMailProfilePolicy: clean_scope = _normalize_scope_type(scope_type) tenant_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="tenant", scope_id=tenant_id) if tenant_policy is None: raise MailProfileError("Tenant not found for mail-server profile policy") policy = EffectiveMailProfilePolicy() system_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="system") or {} _merge_policy(policy, system_policy, source="system", label="System", baseline=True) if clean_scope == "tenant": return policy _merge_policy(policy, tenant_policy, source="tenant", source_id=tenant_id, label="Tenant") if clean_scope in {"user", "group"}: return policy if clean_scope != "campaign" or not scope_id: return policy campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) if campaign.owner_user_id: user_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="user", scope_id=campaign.owner_user_id) if user_policy is not None: _merge_policy(policy, user_policy, source="user", source_id=campaign.owner_user_id, label="Owner user") if campaign.owner_group_id: group_policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="group", scope_id=campaign.owner_group_id) if group_policy is not None: _merge_policy(policy, group_policy, source="group", source_id=campaign.owner_group_id, label="Owner group") return policy def _normalize_match_value(value: str) -> str: return str(value).strip().casefold() def _pattern_matches(value: str, pattern: str) -> bool: clean_pattern = _normalize_match_value(pattern) if clean_pattern == "*": return True return fnmatch.fnmatchcase(value, clean_pattern) def _assert_policy_value(policy: EffectiveMailProfilePolicy, key: str, value: str | None) -> None: allowed, reason = policy.value_allowed(key, value) if not allowed: raise MailProfileError(reason or "Mail profile policy blocked this value") def _domain_from_email(value: str | None) -> str | None: if not value: return None text = str(value).strip().casefold() if "@" not in text: return text or None return text.rsplit("@", 1)[1] or None def _assert_transport_values_allowed(policy: EffectiveMailProfilePolicy, smtp: SmtpConfig | dict[str, Any] | None, imap: ImapConfig | dict[str, Any] | None) -> None: smtp_host = smtp.host if isinstance(smtp, SmtpConfig) else smtp.get("host") if isinstance(smtp, dict) else None imap_host = imap.host if isinstance(imap, ImapConfig) else imap.get("host") if isinstance(imap, dict) else None _assert_policy_value(policy, "smtp_hosts", str(smtp_host) if smtp_host else None) _assert_policy_value(policy, "imap_hosts", str(imap_host) if imap_host else None) def assert_mail_policy_allows_transport( session: Session, *, tenant_id: str, smtp: SmtpConfig | dict[str, Any] | None, imap: ImapConfig | dict[str, Any] | None = None, campaign_id: str | None = None, owner_user_id: str | None = None, owner_group_id: str | None = None, ) -> None: policy = effective_mail_profile_policy( session, tenant_id=tenant_id, campaign_id=campaign_id, owner_user_id=owner_user_id, owner_group_id=owner_group_id, ) _assert_transport_values_allowed(policy, smtp, imap) def assert_mail_policy_allows_send( session: Session, *, tenant_id: str, campaign_id: str, smtp: SmtpConfig | dict[str, Any] | None, imap: ImapConfig | dict[str, Any] | None = None, envelope_sender: str | None = None, from_header: str | None = None, recipients: Iterable[str] = (), ) -> None: policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=campaign_id) _assert_transport_values_allowed(policy, smtp, imap) _assert_policy_value(policy, "envelope_senders", envelope_sender) _assert_policy_value(policy, "from_headers", from_header) for recipient in recipients: _assert_policy_value(policy, "recipient_domains", _domain_from_email(recipient)) def _scope_tuple_for_create( session: Session, *, tenant_id: str, scope_type: str | None, scope_id: str | None, user_id: str | None, ) -> tuple[str | None, str, str | None]: clean_scope = _normalize_scope_type(scope_type) clean_scope_id = str(scope_id).strip() if scope_id else None if clean_scope == "system": return None, clean_scope, None if clean_scope == "tenant": return tenant_id, clean_scope, tenant_id if clean_scope == "user": target_id = clean_scope_id or user_id if not target_id: raise MailProfileError("User-scoped mail-server profiles require a user scope_id") if not _user_exists(session, tenant_id=tenant_id, user_id=target_id): raise MailProfileError("User scope is not part of the active tenant") return tenant_id, clean_scope, target_id if clean_scope == "group": if not clean_scope_id: raise MailProfileError("Group-scoped mail-server profiles require a group scope_id") if not _group_exists(session, tenant_id=tenant_id, group_id=clean_scope_id): raise MailProfileError("Group scope is not part of the active tenant") 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_policy_context(session, tenant_id=tenant_id, campaign_id=clean_scope_id) return tenant_id, clean_scope, clean_scope_id def _ensure_scope_allows_profile_creation( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None, ) -> None: if scope_type in {"system", "tenant"}: return if scope_type == "campaign": policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=scope_id) if not policy.allow_campaign_profiles: raise MailProfileError("Campaign-local mail profiles are disabled by policy") return if scope_type == "user": policy = effective_mail_profile_policy(session, tenant_id=tenant_id, owner_user_id=scope_id) if not policy.allow_user_profiles: raise MailProfileError("User-scoped mail-server profiles are disabled by policy") return if scope_type == "group": policy = effective_mail_profile_policy(session, tenant_id=tenant_id, owner_group_id=scope_id) if not policy.allow_group_profiles: raise MailProfileError("Group-scoped mail-server profiles are disabled by policy") def _ensure_unique_slug( session: Session, *, scope_type: str, scope_id: str | None, slug: str, profile_id: str | None = None, ) -> None: query = session.query(MailServerProfile).filter( MailServerProfile.scope_type == scope_type, MailServerProfile.scope_id.is_(None) if scope_id is None else MailServerProfile.scope_id == scope_id, MailServerProfile.slug == slug, ) if profile_id: query = query.filter(MailServerProfile.id != profile_id) if query.one_or_none() is not None: raise MailProfileError("A mail-server profile with this slug already exists in this scope") def _profile_visible_filter(tenant_id: str): return or_(MailServerProfile.scope_type == "system", MailServerProfile.tenant_id == tenant_id) def _context_profile_clauses(campaign: CampaignMailPolicyContext, policy: EffectiveMailProfilePolicy) -> list[Any]: clauses: list[Any] = [ MailServerProfile.scope_type == "system", and_(MailServerProfile.tenant_id == campaign.tenant_id, MailServerProfile.scope_type == "tenant"), ] if policy.allow_user_profiles and campaign.owner_user_id: clauses.append(and_(MailServerProfile.tenant_id == campaign.tenant_id, MailServerProfile.scope_type == "user", MailServerProfile.scope_id == campaign.owner_user_id)) if policy.allow_group_profiles and campaign.owner_group_id: clauses.append(and_(MailServerProfile.tenant_id == campaign.tenant_id, MailServerProfile.scope_type == "group", MailServerProfile.scope_id == campaign.owner_group_id)) if policy.allow_campaign_profiles: clauses.append(and_(MailServerProfile.tenant_id == campaign.tenant_id, MailServerProfile.scope_type == "campaign", MailServerProfile.scope_id == campaign.id)) return clauses def _profile_transport_allowed(profile: MailServerProfile, policy: EffectiveMailProfilePolicy) -> bool: try: _assert_transport_values_allowed(policy, profile.smtp_config or {}, profile.imap_config or {}) return True except MailProfileError: return False def _profile_allowed_for_context( profile: MailServerProfile, *, tenant_id: str, policy: EffectiveMailProfilePolicy, campaign: CampaignMailPolicyContext | None = None, owner_user_id: str | None = None, owner_group_id: str | None = None, require_active: bool = True, ) -> bool: scope_type = _profile_scope_type(profile) scope_id = _profile_scope_id(profile) if require_active and not profile.is_active: return False if not policy.profile_id_allowed(profile.id): return False if scope_type == "system": pass elif scope_type == "tenant": if profile.tenant_id != tenant_id: return False elif scope_type == "user": if not policy.allow_user_profiles or profile.tenant_id != tenant_id or scope_id != owner_user_id: return False elif scope_type == "group": if not policy.allow_group_profiles or profile.tenant_id != tenant_id or scope_id != owner_group_id: return False elif scope_type == "campaign": if not campaign or not policy.allow_campaign_profiles or profile.tenant_id != tenant_id or scope_id != campaign.id: return False else: return False return _profile_transport_allowed(profile, policy) def campaign_mail_context_visible_to_actor( session: Session, *, tenant_id: str, campaign_id: str, user_id: str, group_ids: Iterable[str] = (), tenant_admin: bool = False, ) -> bool: provider = _campaign_access_provider() if provider is None: return False return bool( provider.can_read_campaign( session, tenant_id=tenant_id, campaign_id=campaign_id, user_id=user_id, group_ids=group_ids, tenant_admin=tenant_admin, ) ) def mail_profile_scope_visible_to_actor( session: Session, *, scope_type: str, scope_id: str | None, profile_tenant_id: str | None, tenant_id: str, user_id: str, group_ids: Iterable[str] = (), can_manage_tenant_profiles: bool = False, tenant_admin: bool = False, ) -> bool: """Apply stable ownership/ACL discovery without activity or policy state.""" clean_scope_type = _normalize_scope_type(scope_type) if clean_scope_type == "system": return True if profile_tenant_id != tenant_id: return False if clean_scope_type == "tenant": return True if clean_scope_type == "user": return bool(can_manage_tenant_profiles or (scope_id and scope_id == user_id)) if clean_scope_type == "group": return bool( can_manage_tenant_profiles or (scope_id and scope_id in {str(group_id) for group_id in group_ids}) ) if clean_scope_type != "campaign" or not scope_id: return False # Mail administration is not a Campaign ACL override. Only Campaign's own # access capability decides this, including its exact tenant-admin rule. return campaign_mail_context_visible_to_actor( session, tenant_id=tenant_id, campaign_id=scope_id, user_id=user_id, group_ids=group_ids, tenant_admin=tenant_admin, ) def mail_profile_visible_to_actor( session: Session, *, profile: MailServerProfile, tenant_id: str, user_id: str, group_ids: Iterable[str] = (), can_manage_own_profiles: bool = False, can_manage_tenant_profiles: bool = False, can_manage_system_profiles: bool = False, tenant_admin: bool = False, administrative_visibility: bool = False, require_active: bool = True, ) -> bool: """Return whether an actor may discover or directly use a profile. System and tenant profiles are shared definitions. Narrower profiles are visible only to their owner context, unless the actor is a tenant-wide Mail profile administrator. In administrative views, a self-service actor may also see their exact user-owned profiles after policy makes them unusable, so they can repair or deactivate them. Campaign visibility is delegated to Campaign's ACL capability so Mail never guesses another module's object permissions. """ normalized_group_ids = tuple(sorted({str(group_id) for group_id in group_ids})) scope_type = _profile_scope_type(profile) scope_id = _profile_scope_id(profile) if not mail_profile_scope_visible_to_actor( session, scope_type=scope_type, scope_id=scope_id, profile_tenant_id=profile.tenant_id, tenant_id=tenant_id, user_id=user_id, group_ids=normalized_group_ids, can_manage_tenant_profiles=( administrative_visibility and can_manage_tenant_profiles ), tenant_admin=tenant_admin, ): return False if administrative_visibility and scope_type == "system" and can_manage_system_profiles: return True if administrative_visibility and scope_type != "system" and can_manage_tenant_profiles: return True if ( administrative_visibility and can_manage_own_profiles and scope_type == "user" and scope_id == user_id ): return True if scope_type in {"system", "tenant"}: return _profile_allowed_for_actor_policy( session, profile=profile, tenant_id=tenant_id, user_id=user_id, group_ids=normalized_group_ids, require_active=require_active, ) if scope_type == "user": return _profile_allowed_for_actor_policy( session, profile=profile, tenant_id=tenant_id, user_id=user_id, group_ids=(), require_active=require_active, ) if scope_type == "group": return _profile_allowed_for_actor_policy( session, profile=profile, tenant_id=tenant_id, user_id=user_id, group_ids=(scope_id,), require_active=require_active, ) if scope_type != "campaign" or not scope_id: return False campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=scope_id) return _profile_allowed_for_context( profile, tenant_id=tenant_id, policy=policy, campaign=campaign, owner_user_id=campaign.owner_user_id, owner_group_id=campaign.owner_group_id, require_active=require_active, ) def _profile_allowed_for_actor_policy( session: Session, *, profile: MailServerProfile, tenant_id: str, user_id: str, group_ids: Iterable[str], require_active: bool = True, ) -> bool: """Apply every unambiguous actor policy context conservatively. Direct Mail screens have no selected task/group context. The actor's user policy always applies; for shared system/tenant profiles each group policy also has to allow the profile. A caller that needs campaign-specific policy must use the explicit campaign context path instead. """ normalized_group_ids = tuple(sorted({str(group_id) for group_id in group_ids})) scope_type = _profile_scope_type(profile) if scope_type == "group": contexts: tuple[str | None, ...] = normalized_group_ids elif scope_type == "user": contexts = (None,) else: contexts = (None, *normalized_group_ids) if not contexts: return False for group_id in contexts: policy = effective_mail_profile_policy( session, tenant_id=tenant_id, owner_user_id=user_id, owner_group_id=group_id, ) if not _profile_allowed_for_context( profile, tenant_id=tenant_id, policy=policy, owner_user_id=user_id, owner_group_id=group_id, require_active=require_active, ): return False return True def list_mail_server_profiles( session: Session, *, tenant_id: str, include_inactive: bool = False, campaign_id: str | None = None, actor_user_id: str | None = None, actor_group_ids: Iterable[str] = (), actor_can_manage_own_profiles: bool = False, actor_can_manage_tenant_profiles: bool = False, actor_can_manage_system_profiles: bool = False, actor_tenant_admin: bool = False, actor_administrative_visibility: bool = False, ) -> list[MailServerProfile]: normalized_actor_group_ids = tuple( sorted({str(group_id) for group_id in actor_group_ids}) ) if campaign_id: campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_id) if actor_user_id and not campaign_mail_context_visible_to_actor( session, tenant_id=tenant_id, campaign_id=campaign.id, user_id=actor_user_id, group_ids=normalized_actor_group_ids, tenant_admin=actor_tenant_admin, ): raise MailProfileError("Campaign not found for mail-server profile policy") policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=campaign.id) query = session.query(MailServerProfile).filter(or_(*_context_profile_clauses(campaign, policy))) if not include_inactive: query = query.filter(MailServerProfile.is_active.is_(True)) profiles = query.all() profiles = [ profile for profile in profiles if (include_inactive or profile.is_active) and _profile_allowed_for_context( profile, tenant_id=tenant_id, policy=policy, campaign=campaign, owner_user_id=campaign.owner_user_id, owner_group_id=campaign.owner_group_id, require_active=not include_inactive, ) ] else: query = session.query(MailServerProfile).filter(_profile_visible_filter(tenant_id)) if not include_inactive: query = query.filter(MailServerProfile.is_active.is_(True)) profiles = query.all() if actor_user_id: profiles = [ profile for profile in profiles if mail_profile_visible_to_actor( session, profile=profile, tenant_id=tenant_id, user_id=actor_user_id, group_ids=normalized_actor_group_ids, can_manage_own_profiles=actor_can_manage_own_profiles, can_manage_tenant_profiles=actor_can_manage_tenant_profiles, can_manage_system_profiles=actor_can_manage_system_profiles, tenant_admin=actor_tenant_admin, administrative_visibility=actor_administrative_visibility, require_active=not include_inactive, ) ] return sorted(profiles, key=lambda item: (PROFILE_SCOPE_ORDER.get(_profile_scope_type(item), 99), (item.name or "").casefold())) def get_mail_server_profile_for_actor( session: Session, *, tenant_id: str, profile_id: str, user_id: str, group_ids: Iterable[str] = (), can_manage_own_profiles: bool = False, can_manage_tenant_profiles: bool = False, can_manage_system_profiles: bool = False, tenant_admin: bool = False, administrative_visibility: bool = False, require_active: bool = False, ) -> MailServerProfile: profile = get_mail_server_profile( session, tenant_id=tenant_id, profile_id=profile_id, require_active=require_active, ) if not mail_profile_visible_to_actor( session, profile=profile, tenant_id=tenant_id, user_id=user_id, group_ids=group_ids, can_manage_own_profiles=can_manage_own_profiles, can_manage_tenant_profiles=can_manage_tenant_profiles, can_manage_system_profiles=can_manage_system_profiles, tenant_admin=tenant_admin, administrative_visibility=administrative_visibility, require_active=require_active, ): # Deliberately indistinguishable from an absent profile to prevent id # probing across user, group, and campaign scopes. raise MailProfileError("Mail-server profile not found") return profile def get_mail_server_profile( session: Session, *, tenant_id: str, profile_id: str, require_active: bool = False, for_update: bool = False, mutation_owner_user_id: str | None = None, can_manage_tenant_profiles: bool = False, can_manage_system_profiles: bool = False, ) -> MailServerProfile: if for_update: statement = _mail_server_profile_mutation_statement( tenant_id=tenant_id, profile_id=profile_id, owner_user_id=mutation_owner_user_id, can_manage_tenant_profiles=can_manage_tenant_profiles, can_manage_system_profiles=can_manage_system_profiles, ) profile = session.execute(statement).scalar_one_or_none() else: profile = session.get(MailServerProfile, profile_id) if profile is None or (_profile_scope_type(profile) != "system" and profile.tenant_id != tenant_id): raise MailProfileError("Mail-server profile not found") if require_active and not profile.is_active: raise MailProfileError("Mail-server profile is inactive") return profile def _mail_server_profile_mutation_statement( *, tenant_id: str, profile_id: str, owner_user_id: str | None, can_manage_tenant_profiles: bool, can_manage_system_profiles: bool, ): """Build the authorization-bounded row lock used by profile mutations. The selector prevents a self-service actor from locking another owner's row merely by guessing its identifier. ``populate_existing`` is essential: after PostgreSQL waits for a concurrent writer, authorization and secret checks must observe that writer's committed password and active state, not an older identity-map snapshot. """ allowed_scopes = [] if can_manage_system_profiles: allowed_scopes.append( and_( MailServerProfile.scope_type == "system", MailServerProfile.tenant_id.is_(None), ) ) if can_manage_tenant_profiles: allowed_scopes.append( and_( MailServerProfile.tenant_id == tenant_id, MailServerProfile.scope_type != "system", ) ) if owner_user_id: allowed_scopes.append( and_( MailServerProfile.tenant_id == tenant_id, MailServerProfile.scope_type == "user", MailServerProfile.scope_id == owner_user_id, ) ) if not allowed_scopes: # Preserve a non-enumerating not-found result without locking an # unauthorized row. allowed_scopes.append(MailServerProfile.id.is_(None)) return ( select(MailServerProfile) .where( MailServerProfile.id == profile_id, or_(*allowed_scopes), ) .with_for_update() .execution_options(populate_existing=True) ) def ensure_mail_profile_allowed_for_campaign( session: Session, *, tenant_id: str, campaign_id: str, profile_id: str, require_active: bool = True, ) -> MailServerProfile: campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=campaign_id) profile = get_mail_server_profile(session, tenant_id=tenant_id, profile_id=profile_id, require_active=require_active) policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=campaign.id) if not _profile_allowed_for_context( profile, tenant_id=tenant_id, policy=policy, campaign=campaign, owner_user_id=campaign.owner_user_id, owner_group_id=campaign.owner_group_id, require_active=require_active, ): raise MailProfileError("Mail-server profile is not allowed for this campaign owner or policy") return profile def _credential_policy_for_protocol(policy: EffectiveMailProfilePolicy, protocol: str) -> EffectiveCredentialPolicy: if protocol == "smtp": return policy.smtp_credentials if protocol == "imap": return policy.imap_credentials raise MailProfileError("Credential policy protocol must be smtp or imap") def _profile_has_transport(profile: MailServerProfile, protocol: str) -> bool: if protocol == "smtp": return True return bool(profile.imap_config) _CAMPAIGN_MAIL_REFERENCE_KEYS = frozenset({"mail_profile_id"}) def _campaign_mail_profile_reference_id(server: dict[str, Any]) -> str | None: unexpected = sorted(key for key in server if key not in _CAMPAIGN_MAIL_REFERENCE_KEYS) if unexpected: paths = ", ".join(f"server.{key}" for key in unexpected) raise MailProfileError( "Campaign JSON may only reference a Mail-owned profile through server.mail_profile_id; " f"remove campaign-local SMTP/IMAP settings ({paths}), select a Mail profile, and save a new campaign version." ) value = server.get("mail_profile_id") if value is None: return None if not isinstance(value, str) or not value.strip(): raise MailProfileError("server.mail_profile_id must be a non-empty Mail profile identifier") return value.strip() def _assert_campaign_inherits_profile_credentials( profile: MailServerProfile, policy: EffectiveMailProfilePolicy, ) -> None: for protocol in ("smtp", "imap"): if not _profile_has_transport(profile, protocol): continue if not _credential_policy_for_protocol(policy, protocol).inherit: raise MailProfileError( f"Campaign delivery cannot use the selected profile because the effective {protocol.upper()} " "credential policy requires campaign-local credentials. Store the credentials on a Mail profile " "and change the policy to inherit them." ) def assert_campaign_mail_policy_allows_json( session: Session, *, tenant_id: str, raw_json: dict[str, Any] | None, campaign_id: str | None = None, owner_user_id: str | None = None, owner_group_id: str | None = None, ) -> None: data = raw_json if isinstance(raw_json, dict) else {} server = data.get("server") if isinstance(data.get("server"), dict) else {} profile_id = _campaign_mail_profile_reference_id(server) if profile_id: if campaign_id: profile = ensure_mail_profile_allowed_for_campaign( session, tenant_id=tenant_id, campaign_id=campaign_id, profile_id=str(profile_id), require_active=True, ) policy = effective_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=campaign_id) _assert_campaign_inherits_profile_credentials(profile, policy) return profile = get_mail_server_profile(session, tenant_id=tenant_id, profile_id=str(profile_id), require_active=True) policy = effective_mail_profile_policy( session, tenant_id=tenant_id, owner_user_id=owner_user_id, owner_group_id=owner_group_id, ) if not _profile_allowed_for_context( profile, tenant_id=tenant_id, policy=policy, owner_user_id=owner_user_id, owner_group_id=owner_group_id, ): raise MailProfileError("Mail-server profile is not allowed by the effective policy") _assert_campaign_inherits_profile_credentials(profile, policy) return return def create_mail_server_profile( session: Session, *, tenant_id: str, user_id: str | None, name: str, slug: str | None, description: str | None, smtp: SmtpConfig, imap: ImapConfig | None, is_active: bool = True, scope_type: str = "tenant", scope_id: str | None = None, ) -> MailServerProfile: clean_name = name.strip() if not clean_name: raise MailProfileError("Profile name cannot be empty") profile_tenant_id, clean_scope_type, clean_scope_id = _scope_tuple_for_create( session, tenant_id=tenant_id, scope_type=scope_type, scope_id=scope_id, user_id=user_id, ) _ensure_scope_allows_profile_creation(session, tenant_id=tenant_id, scope_type=clean_scope_type, scope_id=clean_scope_id) if clean_scope_type == "campaign": assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=smtp, imap=imap, campaign_id=clean_scope_id) elif clean_scope_type == "user": assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=smtp, imap=imap, owner_user_id=clean_scope_id) elif clean_scope_type == "group": assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=smtp, imap=imap, owner_group_id=clean_scope_id) elif clean_scope_type == "tenant": 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_username, smtp_password, _, _ = _transport_payload(smtp) imap_payload = None imap_username = None imap_password = None if imap is not None: imap_payload, imap_username, imap_password, _, _ = _transport_payload(imap) if not is_active and (smtp_password not in (None, "") or imap_password not in (None, "")): raise MailProfileError( "Inactive mail-server profiles cannot retain credentials; create the active profile or omit credentials" ) profile = MailServerProfile( tenant_id=profile_tenant_id, scope_type=clean_scope_type, scope_id=clean_scope_id, name=clean_name, slug=clean_slug, 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, ) session.add(profile) session.flush() return profile def update_mail_server_profile( session: Session, profile: MailServerProfile, *, user_id: str | None, api_key_id: str | None = None, tenant_id: str, name: str | None = None, slug: str | None = None, description: str | None = None, is_active: bool | None = None, smtp: SmtpConfig | None = None, imap: ImapConfig | None = None, clear_imap: bool = False, ) -> MailServerProfile: if is_active is False: raise MailProfileError( "Deactivate profiles through DELETE so credentials are scrubbed and the deletion is audited" ) if not profile.is_active and ( profile.smtp_password_encrypted or profile.imap_password_encrypted ): raise MailProfileError( "Inactive profile still contains legacy credentials; use DELETE to scrub them before editing or reactivating" ) inactive_password_update = ( not profile.is_active and is_active is not True and any( config is not None and "password" in config.model_fields_set and config.password not in (None, "") for config in (smtp, imap) ) ) if inactive_password_update: raise MailProfileError( "Credentials may only be added to an inactive profile when it is activated in the same update" ) if name is not None: clean_name = name.strip() if not clean_name: raise MailProfileError("Profile name cannot be empty") profile.name = clean_name if slug is not None: clean_slug = slugify_profile_name(slug) _ensure_unique_slug( session, scope_type=_profile_scope_type(profile), scope_id=_profile_scope_id(profile), slug=clean_slug, profile_id=profile.id, ) profile.slug = clean_slug if description is not None: profile.description = description if is_active is not None: profile.is_active = is_active next_smtp, next_imap = _next_profile_transport_state(profile, smtp=smtp, imap=imap, clear_imap=clear_imap) _assert_profile_transport_allowed(session, tenant_id=tenant_id, profile=profile, smtp=next_smtp, imap=next_imap) _apply_profile_transport_update( session, profile, user_id=user_id, api_key_id=api_key_id, smtp=smtp, imap=imap, clear_imap=clear_imap, ) profile.updated_by_user_id = user_id session.add(profile) session.flush() return profile def _next_profile_transport_state( profile: MailServerProfile, *, smtp: SmtpConfig | None, imap: ImapConfig | None, clear_imap: bool, ) -> tuple[SmtpConfig | dict[str, Any], ImapConfig | dict[str, Any] | None]: # Policy validation only needs non-secret transport metadata. Never decrypt # an unchanged protocol merely because profile metadata or the other # protocol is being updated; corrupt/retired ciphertext must not block the # credential replacement and deletion recovery paths. next_smtp: SmtpConfig | dict[str, Any] = smtp or dict(profile.smtp_config or {}) if clear_imap: return next_smtp, None if imap is not None: return next_smtp, imap return next_smtp, dict(profile.imap_config or {}) if profile.imap_config else None def _assert_profile_transport_allowed( session: Session, *, tenant_id: str, profile: MailServerProfile, smtp: SmtpConfig | dict[str, Any], imap: ImapConfig | dict[str, Any] | None, ) -> None: scope_type = _profile_scope_type(profile) scope_id = _profile_scope_id(profile) if scope_type == "campaign": assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=smtp, imap=imap, campaign_id=scope_id) elif scope_type == "user": assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=smtp, imap=imap, owner_user_id=scope_id) elif scope_type == "group": assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=smtp, imap=imap, owner_group_id=scope_id) elif scope_type == "tenant": assert_mail_policy_allows_transport(session, tenant_id=tenant_id, smtp=smtp, imap=imap) def _apply_profile_transport_update( session: Session, profile: MailServerProfile, *, user_id: str | None, api_key_id: str | None, smtp: SmtpConfig | None, imap: ImapConfig | None, clear_imap: bool, ) -> None: if smtp is not None: _apply_profile_transport_payload( session, profile, "smtp", smtp, user_id=user_id, api_key_id=api_key_id, ) if clear_imap: had_imap_identity = bool( profile.imap_config or profile.imap_username or profile.imap_password_encrypted ) _audit_profile_credential_mutation( session, profile=profile, protocol="imap", password_supplied=True, next_password=None, user_id=user_id, api_key_id=api_key_id, ) clear_mailbox_index(session, profile_id=profile.id) profile.imap_config = None profile.imap_username = None profile.imap_password_encrypted = None if had_imap_identity: profile.imap_transport_revision = new_uuid() elif imap is not None: _apply_profile_transport_payload( session, profile, "imap", imap, user_id=user_id, api_key_id=api_key_id, ) def _apply_profile_transport_payload( session: Session, profile: MailServerProfile, protocol: str, config: SmtpConfig | ImapConfig, *, user_id: str | None, api_key_id: str | None, ) -> None: payload, username, password, username_supplied, password_supplied = _transport_payload(config) current_payload = dict(profile.smtp_config or {}) if protocol == "smtp" else dict(profile.imap_config or {}) for field_name in ("username", "password", "enabled"): current_payload.pop(field_name, None) current_username = _profile_username(profile, protocol) identity_changed = ( current_payload != payload or (username_supplied and current_username != username) ) _audit_profile_credential_mutation( session, profile=profile, protocol=protocol, password_supplied=password_supplied, next_password=password, user_id=user_id, api_key_id=api_key_id, ) if protocol == "smtp": profile.smtp_config = payload if username_supplied: profile.smtp_username = username if password_supplied: profile.smtp_password_encrypted = encrypt_secret(password) if identity_changed: profile.smtp_transport_revision = new_uuid() return # Cached mailbox metadata and message summaries belong to the exact IMAP # account identity. Clear them in the same transaction before changing an # identity or credential so another account can never inherit them. if identity_changed or password_supplied: clear_mailbox_index(session, profile_id=profile.id) profile.imap_config = payload if username_supplied: profile.imap_username = username if password_supplied: profile.imap_password_encrypted = encrypt_secret(password) if identity_changed: profile.imap_transport_revision = new_uuid() def _audit_profile_credential_mutation( session: Session, *, profile: MailServerProfile, protocol: str, password_supplied: bool, next_password: str | None, user_id: str | None, api_key_id: str | None, ) -> None: encrypted = ( profile.smtp_password_encrypted if protocol == "smtp" else profile.imap_password_encrypted ) if not password_supplied or not encrypted: return replacing = next_password not in (None, "") audit_event( session, tenant_id=profile.tenant_id, scope="system" if profile.scope_type == "system" else "tenant", user_id=user_id, api_key_id=api_key_id, action=( "mail.profile_credentials_replaced" if replacing else "mail.profile_credentials_deleted" ), object_type="mail_server_profile", object_id=profile.id, details={ "scope_type": profile.scope_type, "scope_id": profile.scope_id, "protocol": protocol, "reason": "profile_updated", }, ) # Force the non-secret audit record to durable storage before replacing or # clearing the in-memory ciphertext. The surrounding transaction keeps the # two changes atomic at commit time. session.flush() 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) 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) def campaign_profile_transport_revisions(profile: MailServerProfile) -> dict[str, str | None]: """Return Mail-owned opaque revisions without exposing transport identity. Revisions are random persisted values, backfilled by migration and rotated whenever the corresponding transport or account identity changes. They cannot be dictionary-tested to recover a host or account name. """ smtp_revision = str(getattr(profile, "smtp_transport_revision", "") or "") if not smtp_revision: raise MailProfileError("Mail profile has no SMTP transport revision; run the Mail database migrations") imap_revision = None if profile.imap_config: imap_revision = str(getattr(profile, "imap_transport_revision", "") or "") if not imap_revision: raise MailProfileError("Mail profile has no IMAP transport revision; run the Mail database migrations") return {"smtp": smtp_revision, "imap": imap_revision} def delete_mail_profile_credentials( session: Session, *, profile: MailServerProfile, deletion_reason: str, user_id: str | None = None, api_key_id: str | None = None, ) -> tuple[str, ...]: """Immediately scrub Mail-owned profile secrets and record the deletion. The audit record contains only protocol names and non-secret profile scope metadata. Any audit/storage failure aborts the caller's transaction, so a profile cannot appear deleted while its encrypted passwords remain. """ deleted_protocols = tuple( protocol for protocol, encrypted in ( ("smtp", profile.smtp_password_encrypted), ("imap", profile.imap_password_encrypted), ) if encrypted ) if not deleted_protocols: return () audit_event( session, tenant_id=profile.tenant_id, scope="system" if profile.scope_type == "system" else "tenant", user_id=user_id, api_key_id=api_key_id, action="mail.profile_credentials_deleted", object_type="mail_server_profile", object_id=profile.id, details={ "scope_type": profile.scope_type, "scope_id": profile.scope_id, "deleted_protocols": list(deleted_protocols), "deletion_reason": deletion_reason, }, ) # Fail before mutating the ORM object if the audit record cannot be stored. session.flush() profile.smtp_password_encrypted = None profile.imap_password_encrypted = None session.add(profile) session.flush() return deleted_protocols def delete_mail_profile_credentials_for_retirement(session: Session) -> int: """Scrub and audit every remaining Mail-owned secret before table drop.""" profiles = ( session.query(MailServerProfile) .filter( or_( MailServerProfile.smtp_password_encrypted.is_not(None), MailServerProfile.imap_password_encrypted.is_not(None), ) ) .order_by(MailServerProfile.id.asc()) .with_for_update() .populate_existing() .all() ) deleted = 0 for profile in profiles: deleted += len( delete_mail_profile_credentials( session, profile=profile, deletion_reason="module_data_retired", ) ) session.flush() return deleted 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, "tenant_id": profile.tenant_id, "scope_type": _profile_scope_type(profile), "scope_id": _profile_scope_id(profile), "name": profile.name, "slug": profile.slug, "description": profile.description, "is_active": profile.is_active, "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, "updated_at": profile.updated_at, } def get_mail_profile_policy( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None, ) -> dict[str, Any]: clean_scope = _normalize_scope_type(scope_type) if clean_scope == "system": return _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="system") or {} if clean_scope == "tenant": policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="tenant", scope_id=scope_id or tenant_id) if policy is None: raise MailProfileError("Tenant policy not found") return policy if not scope_id: raise MailProfileError(f"{clean_scope.capitalize()} mail profile policy requires scope_id") if clean_scope == "user": policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="user", scope_id=scope_id) if policy is None: raise MailProfileError("User policy not found") return policy if clean_scope == "group": policy = _read_mail_profile_policy(session, tenant_id=tenant_id, scope_type="group", scope_id=scope_id) if policy is None: raise MailProfileError("Group policy not found") return policy campaign = _campaign_policy_context(session, tenant_id=tenant_id, campaign_id=scope_id) return _policy_from_attr(campaign.mail_profile_policy) def _validate_policy_against_parent(parent: EffectiveMailProfilePolicy, normalized: dict[str, Any]) -> None: violations = _policy_parent_lock_violations(parent.allow_lower_level_limits, normalized) if violations: raise MailProfileError(_policy_parent_lock_message(violations[0])) def _locked_boolean_policy_fields( parent_limits: dict[str, bool], normalized: dict[str, Any], ) -> list[str]: return [ key for key in ( "allow_user_profiles", "allow_group_profiles", "allow_campaign_profiles", ) if isinstance(normalized.get(key), bool) and not _policy_limit_enabled(parent_limits, key) ] def _locked_pattern_policy_fields( parent_limits: dict[str, bool], normalized: dict[str, Any], ) -> list[str]: violations: list[str] = [] for kind in ("whitelist", "blacklist"): rules = normalized.get(kind) or {} if not isinstance(rules, dict): continue for key in PROFILE_PATTERN_KEYS: limit_key = f"{kind}.{key}" if _clean_string_list(rules.get(key, [])) and not _policy_limit_enabled( parent_limits, limit_key, ): violations.append(limit_key) return violations def _locked_credential_policy_fields( parent_limits: dict[str, bool], normalized: dict[str, Any], ) -> list[str]: violations: list[str] = [] for protocol in ("smtp", "imap"): local = normalized.get(f"{protocol}_credentials") or {} local_inherit = local.get("inherit") limit_key = f"{protocol}_credentials.inherit" if isinstance(local_inherit, bool) and not _policy_limit_enabled( parent_limits, limit_key, ): violations.append(limit_key) return violations def _locked_lower_limit_policy_fields( parent_limits: dict[str, bool], normalized: dict[str, Any], ) -> list[str]: local_lower_limits = normalized.get("allow_lower_level_limits") or {} if not isinstance(local_lower_limits, dict): return [] return [ f"allow_lower_level_limits.{key}" for key, allowed in local_lower_limits.items() if allowed is True and not _policy_limit_enabled(parent_limits, key) ] def _policy_parent_lock_violations(parent_limits: dict[str, bool], normalized: dict[str, Any]) -> list[str]: violations: list[str] = [] allowed_ids = _meaningful_allow_patterns(normalized.get("allowed_profile_ids") or []) if allowed_ids and not _policy_limit_enabled(parent_limits, "allowed_profile_ids"): violations.append("allowed_profile_ids") violations.extend(_locked_boolean_policy_fields(parent_limits, normalized)) violations.extend(_locked_pattern_policy_fields(parent_limits, normalized)) violations.extend(_locked_credential_policy_fields(parent_limits, normalized)) violations.extend(_locked_lower_limit_policy_fields(parent_limits, normalized)) return violations def _policy_parent_lock_message(field: str) -> str: if field == "allowed_profile_ids": return "Mail profile allow-list is locked by an ancestor policy" if field.startswith("allow_lower_level_limits."): key = field.removeprefix("allow_lower_level_limits.") return f"{key} lower-level limiting cannot be re-enabled below a locked ancestor policy" if field.endswith("_credentials.inherit"): protocol = field.split("_", 1)[0].upper() return f"{protocol} credential inheritance is locked by an ancestor policy" return f"{field} is locked by an ancestor policy" def set_mail_profile_policy( session: Session, *, tenant_id: str, scope_type: str, scope_id: str | None = None, policy: dict[str, Any] | None = None, ) -> dict[str, Any]: clean_scope = _normalize_scope_type(scope_type) normalized = normalize_mail_profile_policy(policy or {}) if clean_scope != "system": _validate_policy_against_parent( parent_mail_profile_policy(session, tenant_id=tenant_id, scope_type=clean_scope, scope_id=scope_id or (tenant_id if clean_scope == "tenant" else None)), normalized, ) if clean_scope == "system": return _write_mail_profile_policy(session, tenant_id=tenant_id, scope_type="system", scope_id=None, policy=normalized) if clean_scope == "tenant": if (scope_id and scope_id != tenant_id) or not _tenant_exists(session, tenant_id): raise MailProfileError("Tenant policy not found") return _write_mail_profile_policy(session, tenant_id=tenant_id, scope_type="tenant", scope_id=tenant_id, policy=normalized) if not scope_id: raise MailProfileError(f"{clean_scope.capitalize()} mail profile policy requires scope_id") if clean_scope == "user": if not _user_exists(session, tenant_id=tenant_id, user_id=scope_id): raise MailProfileError("User policy not found") return _write_mail_profile_policy(session, tenant_id=tenant_id, scope_type="user", scope_id=scope_id, policy=normalized) if clean_scope == "group": if not _group_exists(session, tenant_id=tenant_id, group_id=scope_id): raise MailProfileError("Group policy not found") return _write_mail_profile_policy(session, tenant_id=tenant_id, scope_type="group", scope_id=scope_id, policy=normalized) provider = _campaign_policy_provider() if provider is None: raise MailProfileError("Campaign module is not available") try: provider.set_campaign_mail_profile_policy(session, tenant_id=tenant_id, campaign_id=scope_id, policy=normalized) except ValueError as exc: raise MailProfileError("Campaign policy not found") from exc return normalized def mail_profile_id_from_campaign_json(raw_json: dict[str, Any] | None) -> str | None: server = raw_json.get("server") if isinstance(raw_json, dict) else None return _campaign_mail_profile_reference_id(server) if isinstance(server, dict) else None def group_ids_for_user(session: Session, *, tenant_id: str, user_id: str) -> list[str]: directory = _access_directory() if directory is not None: return [group.id for group in directory.groups_for_user(user_id, tenant_id=tenant_id)] rows = session.execute( text("select group_id from access_user_group_memberships where tenant_id = :tenant_id and user_id = :user_id"), {"tenant_id": tenant_id, "user_id": user_id}, ).all() return [row[0] for row in rows]