from __future__ import annotations from typing import Any from sqlalchemy.orm import Session from govoplan_core.core.modules import DocumentationCondition, DocumentationContext, DocumentationLink, DocumentationTopic from govoplan_mail.backend.mail_profiles import ( EffectiveMailProfilePolicy, MailProfileError, effective_mail_profile_policy_for_scope, ) MAIL_POLICY_DOC_SCOPES = ("mail:profile:read", "admin:policies:read", "system:settings:read") MAIL_PROFILE_READ_SCOPE = "mail:profile:read" MAIL_PROFILE_WRITE_SCOPE = "mail:profile:write" MAIL_PROFILE_WRITE_OWN_SCOPE = "mail:profile:write_own" MAIL_SECRET_MANAGE_SCOPE = "mail:secret:manage" # noqa: S105 -- permission identifier, not a credential MAIL_SECRET_MANAGE_OWN_SCOPE = "mail:secret:manage_own" # noqa: S105 -- permission identifier, not a credential MAIL_PROFILE_TEST_SCOPE = "mail:profile:test" MAIL_PROFILE_USE_SCOPE = "mail:profile:use" _HOST_POLICY_FIELDS = (("SMTP", "smtp_hosts"), ("IMAP", "imap_hosts")) def documentation_topics(context: DocumentationContext) -> tuple[DocumentationTopic, ...]: topics = [ topic for topic in ( _tenant_mail_policy_topic(context), _custom_mail_profile_topic(context), ) if topic is not None ] return tuple(topics) def _tenant_mail_policy_topic(context: DocumentationContext) -> DocumentationTopic | None: principal = context.principal tenant_id = str(getattr(principal, "tenant_id", "") or "") if not tenant_id: return None if context.documentation_type == "admin" and not _has_any_scope(principal, MAIL_POLICY_DOC_SCOPES): return None session = context.session if not isinstance(session, Session): return None try: policy = effective_mail_profile_policy_for_scope(session, tenant_id=tenant_id, scope_type="tenant") except MailProfileError as exc: user_documentation = context.documentation_type == "user" return DocumentationTopic( id="mail.tenant-profile-policy-unavailable", title="Mail server policy could not be evaluated", summary="Mail profile documentation is installed, but the tenant-level effective policy could not be read for this request.", body=( "The current Mail policy could not be loaded. Try again or ask a Mail administrator for help." if user_documentation else str(exc) ), layer="available", documentation_types=(context.documentation_type,), source_module_id="mail", order=41, links=( (DocumentationLink(label="Public mail help", href="https://govplan.add-ideas.de/modules/mail", kind="public"),) if user_documentation else (_mail_policy_api_link(),) ), metadata={} if user_documentation else {"error_type": type(exc).__name__}, ) effective = policy.as_dict() if context.documentation_type == "user": summary, body = _mail_policy_user_text(effective) return DocumentationTopic( id="mail.tenant-profile-policy-user", title="Choosing a mail server", summary=summary, body=body, layer="configured", documentation_types=("user",), audience=("mail_user", "campaign_user"), order=40, links=( DocumentationLink(label="Public mail help", href="https://govplan.add-ideas.de/modules/mail", kind="public"), ), related_modules=("campaigns",), unlocks=("Campaigns can use these mail choices when the campaign module is available.",), configuration_keys=("mail_profile_policy",), i18n_key="mail.topic.tenant_profile_policy_user", translations=_user_mail_policy_translations(effective), source_module_id="mail", metadata=_mail_policy_user_metadata(effective), ) summary, body = _mail_policy_admin_text(effective, source_count=len(policy.source_policies)) return DocumentationTopic( id="mail.tenant-profile-policy-admin", title="Mail server choices for this tenant", summary=summary, body=body, layer="configured", documentation_types=("admin",), audience=("tenant_admin", "mail_admin", "campaign_admin"), order=40, links=( DocumentationLink(label="Mail profiles API", href="/api/v1/mail/profiles", kind="api"), _mail_policy_api_link(), DocumentationLink(label="Public mail module documentation", href="https://govplan.add-ideas.de/modules/mail", kind="public"), ), related_modules=("campaigns",), unlocks=("Campaign delivery can use reusable mail profiles when govoplan-campaign is installed.",), configuration_keys=("mail_profile_policy",), i18n_key="mail.topic.tenant_profile_policy_admin", source_module_id="mail", metadata=_mail_policy_metadata(effective, source_count=len(policy.source_policies)), ) def _custom_mail_profile_topic(context: DocumentationContext) -> DocumentationTopic | None: if context.documentation_type != "user": return None principal = context.principal if not _has_any_scope(principal, (MAIL_PROFILE_WRITE_SCOPE, MAIL_PROFILE_WRITE_OWN_SCOPE)): return None if not _has_all_scopes(principal, (MAIL_PROFILE_READ_SCOPE,)): return None tenant_id = str(getattr(principal, "tenant_id", "") or "") user_id = str(getattr(getattr(principal, "user", None), "id", "") or "") session = context.session if not tenant_id or not user_id or not isinstance(session, Session): return None try: policy = effective_mail_profile_policy_for_scope( session, tenant_id=tenant_id, scope_type="user", scope_id=user_id, ) except MailProfileError: # A user-facing runtime topic must not turn internal policy resolution # details into documentation output. The task simply remains absent # until its effective policy can be proven. return None if not policy.allow_user_profiles: return None can_manage_credentials = _has_any_scope( principal, (MAIL_SECRET_MANAGE_SCOPE, MAIL_SECRET_MANAGE_OWN_SCOPE), ) can_test_profile = _has_all_scopes(principal, (MAIL_PROFILE_TEST_SCOPE, MAIL_PROFILE_USE_SCOPE)) can_use_profile = _has_any_scope(principal, (MAIL_PROFILE_USE_SCOPE,)) approval_required = bool(policy.allowed_profile_id_sets) constraints = _host_policy_constraint_records(policy) authority_lines = _custom_profile_authority_lines( can_manage_credentials=can_manage_credentials, can_test_profile=can_test_profile, can_use_profile=can_use_profile, approval_required=approval_required, ) steps = _custom_profile_steps( can_manage_credentials=can_manage_credentials, can_test_profile=can_test_profile, can_use_profile=can_use_profile, approval_required=approval_required, ) return DocumentationTopic( id="mail.workflow.create-custom-profile", title="Create a custom Mail profile", summary=( "Create a reusable profile in the current account's user-scoped Settings view, " "within the active SMTP and IMAP hostname policy." ), body="\n".join(authority_lines), layer="configured", documentation_types=("user",), audience=("mail_profile_author", "campaign_manager"), order=41, conditions=( DocumentationCondition( required_modules=("mail",), required_scopes=(MAIL_PROFILE_READ_SCOPE,), any_scopes=(MAIL_PROFILE_WRITE_SCOPE, MAIL_PROFILE_WRITE_OWN_SCOPE), configuration_keys=("mail_profile_policy",), ), ), links=( DocumentationLink(label="My Mail profiles", href="/settings?section=mail-profiles", kind="runtime"), DocumentationLink(label="Public mail help", href="https://govplan.add-ideas.de/modules/mail", kind="public"), ), related_modules=("campaigns",), unlocks=("A custom Mail-owned transport definition that authorized tasks can reference after all policy checks pass.",), configuration_keys=("mail_profile_policy",), i18n_key="mail.topic.create_custom_profile", source_module_id="mail", metadata={ "kind": "workflow", "route": "/settings?section=mail-profiles", "screen": "My Mail profiles", "help_contexts": ["mail.profiles", "app.settings"], "prerequisites": [ "The Mail profile editor is available in Settings and opens in the current account's user scope.", "The effective Mail policy permits user-scoped profiles.", ], "steps": list(steps), "outcome": "A user-scoped custom Mail profile is saved without copying its credentials into a consuming module.", "current_configuration": list(authority_lines), "constraints": list(constraints), "verification": _custom_profile_verification( can_test_profile=can_test_profile, can_use_profile=can_use_profile, approval_required=approval_required, ), "can_manage_credentials": can_manage_credentials, "can_test_profile": can_test_profile, "can_use_profile": can_use_profile, "approval_required_before_use": approval_required, "related_topic_ids": [ "mail.workflow.choose-and-test-profile", "mail.profile-ownership-and-consumers", "mail.profiles-and-policy", ], }, ) def _host_policy_constraint_records(policy: EffectiveMailProfilePolicy) -> tuple[dict[str, Any], ...]: constraints: list[dict[str, Any]] = [] for label, key in _HOST_POLICY_FIELDS: prefix = label.casefold() denied = _display_patterns(policy.blacklist_patterns.get(key, [])) constraints.append({ "id": f"{prefix}-host-deny", "label": f"{label} denied hosts", "description": ( "Deny rules are checked first. The hostname must not match any listed pattern." if denied else "Deny rules are checked first. No hostname deny pattern is active." ), **({"values": list(denied)} if denied else {}), }) allowed_groups = tuple( group for group in (_display_patterns(items) for items in policy.whitelist_groups.get(key, [])) if group ) if not allowed_groups: constraints.append({ "id": f"{prefix}-host-allow", "label": f"{label} allowed hosts", "description": "After deny checks, no hostname allow-list group is active.", }) continue for index, group in enumerate(allowed_groups, start=1): constraints.append({ "id": f"{prefix}-host-allow-{index}", "label": f"{label} allowed hosts — group {index}", "description": ( "After deny checks, the hostname must match at least one pattern in this group. " "It must satisfy every active allow-list group shown for this protocol." ), "values": list(group), }) return tuple(constraints) def _display_patterns(patterns: list[str]) -> tuple[str, ...]: result: list[str] = [] for pattern in patterns: value = str(pattern).strip() if value and value not in result: result.append(value) return tuple(result) def _custom_profile_authority_lines( *, can_manage_credentials: bool, can_test_profile: bool, can_use_profile: bool, approval_required: bool, ) -> tuple[str, ...]: credentials = ( "Credential authority: you may save or replace Mail-owned SMTP/IMAP passwords." if can_manage_credentials else "Credential authority: you may define the profile, but you cannot save or replace passwords; an actor with both profile-write and secret-management authority must do that when authentication requires one." ) testing = ( "Test authority: you may run the profile's SMTP/IMAP connection tests after saving it as active." if can_test_profile else "Test authority: creating the profile does not let you run connection tests; ask an actor with both profile-test and profile-use authority to verify an active profile." ) use = ( "Use authority: you may select the profile in an authorized task after its contextual policy checks pass." if can_use_profile else "Use authority: creating the profile does not let you select or use it; separate Mail profile use authority is required." ) approval = ( "Approval: an approved-profile list is active. A newly generated profile reference must be approved before the profile can be selected or used." if approval_required else "Approval: no approved-profile list currently blocks a newly created profile, but each consuming task still rechecks its contextual policy." ) return credentials, testing, use, approval def _custom_profile_steps( *, can_manage_credentials: bool, can_test_profile: bool, can_use_profile: bool, approval_required: bool, ) -> tuple[str, ...]: steps = [ "Open Settings, choose Mail profiles, and select Add profile in the current account's user-scoped view.", "Enter a stable name and configure SMTP plus optional IMAP hostnames that satisfy every host-policy statement shown above.", ] steps.append( "Enter the required SMTP/IMAP credentials in the dedicated password fields." if can_manage_credentials else "Save the non-secret profile definition, then ask an actor with both profile-write and secret-management authority to add credentials if the server requires authentication." ) steps.append("Save the profile; Mail validates the effective user-scope host policy again on the server.") steps.append( "Save the profile as active, then run the available SMTP and IMAP connection tests." if can_test_profile else "Ask an actor with both profile-test and profile-use authority to run the SMTP and IMAP connection tests after the profile is active." ) if approval_required: steps.append("Ask a Mail administrator to add the new profile to the active approved-profile list.") steps.append( "Select the profile from the consuming task's picker and complete that task's contextual validation." if can_use_profile else "Ask an actor with Mail profile use authority to select it in the consuming task after approval and testing." ) return tuple(steps) def _custom_profile_verification(*, can_test_profile: bool, can_use_profile: bool, approval_required: bool) -> str: checks = ["Reopen My Mail profiles and confirm the saved profile remains in the current account's user-scoped view."] checks.append( "Confirm the authorized SMTP/IMAP tests succeed." if can_test_profile else "Have an authorized tester confirm the SMTP/IMAP tests succeed." ) if approval_required: checks.append("Confirm an administrator approved the generated profile reference before expecting it in a picker.") checks.append( "Confirm the intended task can select it and passes its own policy validation." if can_use_profile else "Confirm a separately authorized user can select it and passes the consuming task's policy validation." ) return " ".join(checks) def _mail_policy_admin_text(policy: dict[str, Any], *, source_count: int) -> tuple[str, str]: allowed_profile_ids = policy.get("allowed_profile_ids") lower_scopes = _allowed_lower_scopes(policy) approved_profile_limit = isinstance(allowed_profile_ids, list) pattern_counts = _pattern_counts(policy) locked_limit_count = sum(1 for value in _lower_limit_values(policy) if value is False) if approved_profile_limit and not lower_scopes: summary = "This tenant is in approved-profile mode: users can choose configured mail profiles, but lower scopes cannot bring arbitrary SMTP or IMAP servers." elif approved_profile_limit: summary = "This tenant limits mail sending to approved profile ids, while selected lower scopes can still define profiles within policy limits." elif lower_scopes: summary = "This tenant permits lower-scope mail profiles subject to the effective host, sender, recipient, and credential policy." else: summary = "This tenant uses centrally managed mail profiles; lower-scope profile creation is disabled by the effective policy." approved_line = _approved_profile_line(allowed_profile_ids) lower_scope_line = _lower_scope_line(lower_scopes) credential_line = _credential_line(policy) pattern_line = f"Allow-list pattern groups: {pattern_counts['whitelist']}. Deny-list patterns: {pattern_counts['blacklist']}." lower_limit_line = f"Locked lower-level limits: {locked_limit_count}. Policy sources applied: {source_count}." body = "\n".join([approved_line, lower_scope_line, credential_line, pattern_line, lower_limit_line]) return summary, body def _mail_policy_user_text(policy: dict[str, Any]) -> tuple[str, str]: allowed_profile_ids = policy.get("allowed_profile_ids") lower_scopes = _allowed_lower_scopes(policy) approved_profile_limit = isinstance(allowed_profile_ids, list) if approved_profile_limit and not lower_scopes: summary = "You can choose from approved mail servers, but you cannot add your own mail server for this tenant." body = "This is set by tenant policy. If the mail server you need is not offered, ask an administrator to add it as an approved mail profile." elif approved_profile_limit: summary = "You can use approved Mail profiles. Some scopes may also define additional reusable profiles." body = "The available profiles are limited by tenant policy. Campaigns select one profile by reference; SMTP/IMAP settings and credentials remain managed in Mail." elif lower_scopes: summary = "You may be able to define reusable Mail profiles in selected scopes, subject to tenant rules." body = "GovOPlaN checks each profile before use. Campaigns reference an available profile and never store its SMTP/IMAP settings or credentials." else: summary = "Mail servers are managed centrally for this tenant." body = "You cannot add a personal, group, or campaign mail server here. Choose one of the configured options or ask an administrator to add another approved profile." return summary, body def _user_mail_policy_translations(policy: dict[str, Any]) -> dict[str, dict[str, str]]: allowed_profile_ids = policy.get("allowed_profile_ids") lower_scopes = _allowed_lower_scopes(policy) approved_profile_limit = isinstance(allowed_profile_ids, list) if approved_profile_limit and not lower_scopes: return { "de": { "title": "Mailserver auswaehlen", "summary": "Sie koennen aus freigegebenen Mailservern waehlen, aber keinen eigenen Mailserver fuer diesen Tenant eintragen.", "body": "Das wird durch die Tenant-Regeln festgelegt. Wenn der benoetigte Mailserver nicht angeboten wird, bitten Sie eine Administratorin oder einen Administrator, ihn als freigegebenes Mailprofil anzulegen.", }, } if approved_profile_limit: return { "de": { "title": "Mailserver auswaehlen", "summary": "Sie koennen freigegebene Mailprofile nutzen. In manchen Bereichen koennen weitere wiederverwendbare Profile angelegt werden.", "body": "Die Auswahl wird durch Tenant-Regeln begrenzt. Kampagnen speichern nur die Profilreferenz; SMTP-/IMAP-Einstellungen und Zugangsdaten verbleiben im Mail-Modul.", }, } if lower_scopes: return { "de": { "title": "Mailserver auswaehlen", "summary": "In bestimmten Bereichen koennen Sie wiederverwendbare Mailprofile anlegen, solange die aktiven Tenant-Regeln eingehalten werden.", "body": "GovOPlaN prueft jedes Profil vor der Verwendung. Kampagnen referenzieren ein freigegebenes Profil und speichern keine SMTP-/IMAP-Einstellungen oder Zugangsdaten.", }, } return { "de": { "title": "Mailserver auswaehlen", "summary": "Mailserver werden fuer diesen Tenant zentral verwaltet.", "body": "Sie koennen hier keinen persoenlichen, Gruppen- oder Kampagnen-Mailserver eintragen. Waehlen Sie eine konfigurierte Option oder bitten Sie eine Administratorin oder einen Administrator um ein weiteres freigegebenes Profil.", }, } def _approved_profile_line(allowed_profile_ids: Any) -> str: if isinstance(allowed_profile_ids, list): count = len(allowed_profile_ids) if count == 0: return "Approved profiles: the active policy currently allows no reusable profile ids." return f"Approved profiles: users are limited to {count} reusable profile id(s) selected by policy." return "Approved profiles: no explicit approved-profile list is active." def _lower_scope_line(lower_scopes: tuple[str, ...]) -> str: if not lower_scopes: return "Lower scopes: user, group, and campaign-local profile creation are disabled." return "Lower scopes allowed to define profiles: " + ", ".join(lower_scopes) + "." def _credential_line(policy: dict[str, Any]) -> str: smtp_inherit = bool((policy.get("smtp_credentials") or {}).get("inherit", True)) imap_inherit = bool((policy.get("imap_credentials") or {}).get("inherit", True)) return ( f"Credential inheritance: SMTP {'inherits' if smtp_inherit else 'requires local credentials'}; " f"IMAP {'inherits' if imap_inherit else 'requires local credentials'}. " "Campaign delivery is available only for protocols that inherit credentials from the selected Mail profile." ) def _allowed_lower_scopes(policy: dict[str, Any]) -> tuple[str, ...]: scopes: list[str] = [] if policy.get("allow_user_profiles") is not False: scopes.append("user") if policy.get("allow_group_profiles") is not False: scopes.append("group") if policy.get("allow_campaign_profiles") is not False: scopes.append("campaign") return tuple(scopes) def _pattern_counts(policy: dict[str, Any]) -> dict[str, int]: return { "whitelist": _pattern_count(policy.get("whitelist")), "blacklist": _pattern_count(policy.get("blacklist")), } def _pattern_count(payload: Any) -> int: if not isinstance(payload, dict): return 0 return sum(len(value) for value in payload.values() if isinstance(value, list)) def _lower_limit_values(policy: dict[str, Any]) -> tuple[bool, ...]: lower_limits = policy.get("allow_lower_level_limits") if not isinstance(lower_limits, dict): return () return tuple(value for value in lower_limits.values() if isinstance(value, bool)) def _mail_policy_metadata(policy: dict[str, Any], *, source_count: int) -> dict[str, Any]: allowed_profile_ids = policy.get("allowed_profile_ids") pattern_counts = _pattern_counts(policy) return { "approved_profile_limit_active": isinstance(allowed_profile_ids, list), "approved_profile_count": len(allowed_profile_ids) if isinstance(allowed_profile_ids, list) else None, "lower_scopes_allowed": list(_allowed_lower_scopes(policy)), "whitelist_pattern_count": pattern_counts["whitelist"], "blacklist_pattern_count": pattern_counts["blacklist"], "locked_lower_level_limit_count": sum(1 for value in _lower_limit_values(policy) if value is False), "policy_source_count": source_count, } def _mail_policy_user_metadata(policy: dict[str, Any]) -> dict[str, Any]: allowed_profile_ids = policy.get("allowed_profile_ids") return { "approved_profile_limit_active": isinstance(allowed_profile_ids, list), "can_add_local_mail_server": bool(_allowed_lower_scopes(policy)), "lower_scopes_allowed": list(_allowed_lower_scopes(policy)), } def _has_any_scope(principal: object | None, scopes: tuple[str, ...]) -> bool: has = getattr(principal, "has", None) if callable(has): return any(bool(has(scope)) for scope in scopes) principal_scopes = set(getattr(principal, "scopes", ()) or ()) return "*" in principal_scopes or any(scope in principal_scopes for scope in scopes) def _has_all_scopes(principal: object | None, scopes: tuple[str, ...]) -> bool: has = getattr(principal, "has", None) if callable(has): try: return all(bool(has(scope)) for scope in scopes) except Exception: return False principal_scopes = set(getattr(principal, "scopes", ()) or ()) return "*" in principal_scopes or all(scope in principal_scopes for scope in scopes) def _mail_policy_api_link() -> DocumentationLink: return DocumentationLink(label="Tenant mail policy API", href="/api/v1/mail/policies/tenant", kind="api")