diff --git a/README.md b/README.md index d493a50..208e9e9 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@ GovOPlaN Mail is the mail transport module. It owns reusable SMTP/IMAP profile m This repository owns: - backend module manifest `mail` -- mail permissions such as `mail:profile:read`, `mail:profile:write`, `mail:profile:use`, `mail:profile:test`, and `mail:mailbox:read` +- mail permissions such as `mail:profile:read`, `mail:profile:write_own`, `mail:profile:write`, `mail:profile:use`, `mail:profile:test`, and `mail:mailbox:read` - SMTP/IMAP profile models, policy checks, encrypted credential storage, and profile resolution - SMTP send and IMAP append adapters, including mock transports for development - development mock mailbox endpoints used by test-send flows @@ -55,6 +55,16 @@ a non-secret audit event. Destructive module retirement applies the same rule to every remaining profile before any Mail table is dropped; a scrub or audit failure blocks retirement. +Personal profile self-service is a distinct authorization path. An actor with +`mail:profile:write_own` can mutate only a user-scoped profile whose scope id is +their current tenant membership id; `mail:secret:manage_own` applies the same +ownership check to credentials. Neither scope permits profile-policy changes or +management of tenant, group, campaign, system, or another user's profiles. +Changing an SMTP/IMAP endpoint while a stored password remains is a secret +operation and requires the matching credential permission. Deactivating a +credential-free profile needs only profile-write authority; if a password will +be scrubbed, credential authority is required and the deletion is audited. + ## Development Install through the core environment: diff --git a/docs/MAIL_HANDBOOK.md b/docs/MAIL_HANDBOOK.md index 283baa6..f5e58c3 100644 --- a/docs/MAIL_HANDBOOK.md +++ b/docs/MAIL_HANDBOOK.md @@ -142,24 +142,38 @@ The supplied templates are: - **Mail profile user:** read/use/test approved profiles and read permitted mailboxes without reading secrets. +- **Mail profile self-service user:** additionally create, edit, deactivate, + and manage credentials only for the current account's own user-scoped + profiles, subject to the effective Mail policy. - **Mail profile administrator:** additionally create/update profiles and - create/replace encrypted credentials. + create/replace encrypted credentials across tenant-owned scopes. The specific permissions are `mail:profile:read`, `mail:profile:use`, -`mail:profile:test`, `mail:mailbox:read`, `mail:profile:write`, and -`mail:secret:manage`. System-scoped definitions use the corresponding system -settings authority. Keep secret management separate when an institution wants -profile metadata administrators not to know or replace credentials. +`mail:profile:test`, `mail:mailbox:read`, `mail:profile:write_own`, +`mail:secret:manage_own`, `mail:profile:write`, and `mail:secret:manage`. +The `_own` permissions are enforced against the authenticated membership id and +never authorize a tenant, group, campaign, system, or another user's profile. +They also do not authorize profile-policy changes. System-scoped definitions +use the corresponding system settings authority. Keep secret management +separate when an institution wants profile metadata administrators not to know +or replace credentials. + +That separation is fail-closed for transport rebinding: changing an SMTP or +IMAP host, port, or security mode while the profile retains a stored password +requires the matching secret-management permission. A credential-free profile +can be deactivated with profile-write authority alone; deactivation that +scrubs a stored password also requires secret-management authority and records +the deletion in the audit log. ### Create or change a profile The configured Help Center exposes **Create a custom Mail profile** only when -the current actor has profile-write authority and the effective user-scope -policy permits user profiles. It states the active SMTP/IMAP hostname +the current actor has broad or self-service profile-write authority and the +effective user-scope policy permits user profiles. It states the active SMTP/IMAP hostname allow-list groups and deny rules, plus the actor's separate credential, test, use, and approval requirements. The Settings task creates in the current -account's user scope; `mail:profile:write` itself remains broad profile -administration authority, not an API-enforced self-only permission. +account's user scope. Grant `mail:profile:write_own` for self-service; +`mail:profile:write` remains broad profile administration authority. 1. Choose the narrowest suitable scope and a stable, descriptive name/slug. 2. Configure SMTP, optional IMAP, TLS mode, account identity, Sent-folder diff --git a/src/govoplan_mail/backend/documentation.py b/src/govoplan_mail/backend/documentation.py index aa33ac1..cc11f86 100644 --- a/src/govoplan_mail/backend/documentation.py +++ b/src/govoplan_mail/backend/documentation.py @@ -14,7 +14,9 @@ from govoplan_mail.backend.mail_profiles import ( 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_SECRET_MANAGE_SCOPE = "mail:secret:manage" +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")) @@ -120,7 +122,9 @@ def _custom_mail_profile_topic(context: DocumentationContext) -> DocumentationTo if context.documentation_type != "user": return None principal = context.principal - if not _has_all_scopes(principal, (MAIL_PROFILE_READ_SCOPE, MAIL_PROFILE_WRITE_SCOPE)): + 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 "") @@ -143,7 +147,10 @@ def _custom_mail_profile_topic(context: DocumentationContext) -> DocumentationTo if not policy.allow_user_profiles: return None - can_manage_credentials = _has_any_scope(principal, (MAIL_SECRET_MANAGE_SCOPE,)) + 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) @@ -175,7 +182,8 @@ def _custom_mail_profile_topic(context: DocumentationContext) -> DocumentationTo conditions=( DocumentationCondition( required_modules=("mail",), - required_scopes=(MAIL_PROFILE_READ_SCOPE, MAIL_PROFILE_WRITE_SCOPE), + required_scopes=(MAIL_PROFILE_READ_SCOPE,), + any_scopes=(MAIL_PROFILE_WRITE_SCOPE, MAIL_PROFILE_WRITE_OWN_SCOPE), configuration_keys=("mail_profile_policy",), ), ), diff --git a/src/govoplan_mail/backend/mail_profiles.py b/src/govoplan_mail/backend/mail_profiles.py index a412075..2507e99 100644 --- a/src/govoplan_mail/backend/mail_profiles.py +++ b/src/govoplan_mail/backend/mail_profiles.py @@ -6,7 +6,7 @@ import re from dataclasses import dataclass, field from typing import Any, Iterable -from sqlalchemy import and_, or_, text +from sqlalchemy import and_, or_, select, text from sqlalchemy.orm import Session from govoplan_core.admin.settings import get_system_settings @@ -968,6 +968,7 @@ def mail_profile_visible_to_actor( 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, @@ -978,8 +979,11 @@ def mail_profile_visible_to_actor( 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. Campaign visibility is delegated to Campaign's ACL - capability so Mail never guesses another module's object permissions. + 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})) @@ -1003,6 +1007,13 @@ def mail_profile_visible_to_actor( 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, @@ -1099,6 +1110,7 @@ def list_mail_server_profiles( 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, @@ -1151,6 +1163,7 @@ def list_mail_server_profiles( 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, @@ -1168,6 +1181,7 @@ def get_mail_server_profile_for_actor( 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, @@ -1186,6 +1200,7 @@ def get_mail_server_profile_for_actor( 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, @@ -1204,8 +1219,22 @@ def get_mail_server_profile( 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: - profile = session.get(MailServerProfile, profile_id) + 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: @@ -1213,6 +1242,61 @@ def get_mail_server_profile( 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, *, @@ -1764,6 +1848,8 @@ def delete_mail_profile_credentials_for_retirement(session: Session) -> int: ) ) .order_by(MailServerProfile.id.asc()) + .with_for_update() + .populate_existing() .all() ) deleted = 0 diff --git a/src/govoplan_mail/backend/manifest.py b/src/govoplan_mail/backend/manifest.py index 0ec71ca..21e1f6d 100644 --- a/src/govoplan_mail/backend/manifest.py +++ b/src/govoplan_mail/backend/manifest.py @@ -80,7 +80,17 @@ PERMISSIONS = ( _permission("mail:profile:test", "Test mail profiles", "Run SMTP/IMAP connection tests."), _permission("mail:mailbox:read", "Read mailboxes", "List IMAP folders and inspect messages without mutating mailbox state."), _permission("mail:profile:write", "Manage mail profiles", "Create and edit reusable mail profiles."), + _permission( + "mail:profile:write_own", + "Manage own mail profiles", + "Create, edit, and deactivate only the current account's user-scoped mail profiles within effective policy.", + ), _permission("mail:secret:manage", "Manage mail secrets", "Create or replace stored SMTP/IMAP credentials."), + _permission( + "mail:secret:manage_own", + "Manage own mail secrets", + "Create, replace, and delete credentials only for the current account's user-scoped mail profiles.", + ), ) ROLE_TEMPLATES = ( @@ -103,6 +113,19 @@ ROLE_TEMPLATES = ( description="Use and test approved mail profiles without reading secrets.", permissions=("mail:profile:read", "mail:profile:use", "mail:profile:test", "mail:mailbox:read"), ), + RoleTemplate( + slug="mail_profile_self_service", + name="Mail profile self-service user", + description="Create and manage only personal Mail profiles and credentials within effective policy.", + permissions=( + "mail:profile:read", + "mail:profile:use", + "mail:profile:test", + "mail:mailbox:read", + "mail:profile:write_own", + "mail:secret:manage_own", + ), + ), ) @@ -126,7 +149,7 @@ def _mail_router(context: ModuleContext): manifest = ModuleManifest( id="mail", name="Mail", - version="0.1.9", + version="0.1.10", required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), optional_dependencies=("campaigns", "addresses"), provides_interfaces=( @@ -231,7 +254,12 @@ manifest = ModuleManifest( conditions=( DocumentationCondition( required_modules=("mail",), - any_scopes=("mail:profile:read", "mail:profile:use", "mail:profile:write"), + any_scopes=( + "mail:profile:read", + "mail:profile:use", + "mail:profile:write", + "mail:profile:write_own", + ), ), ), links=( diff --git a/src/govoplan_mail/backend/router.py b/src/govoplan_mail/backend/router.py index 7793248..a37374d 100644 --- a/src/govoplan_mail/backend/router.py +++ b/src/govoplan_mail/backend/router.py @@ -127,18 +127,133 @@ def _require_any_scope(principal: ApiPrincipal, *scopes: str) -> None: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Requires one of: " + ", ".join(scopes)) -def _require_profile_write_scope(principal: ApiPrincipal, scope_type: str) -> None: - if scope_type == "system": - _require_scope(principal, "system:settings:write") - else: - _require_scope(principal, "mail:profile:write") +def _is_own_user_scope(principal: ApiPrincipal, scope_type: str, scope_id: str | None) -> bool: + return scope_type == "user" and scope_id == principal.user.id -def _require_profile_credentials_scope(principal: ApiPrincipal, scope_type: str) -> None: +def _require_profile_write_scope( + principal: ApiPrincipal, + scope_type: str, + scope_id: str | None = None, +) -> None: if scope_type == "system": _require_scope(principal, "system:settings:write") - else: - _require_scope(principal, "mail:secret:manage") + return + if has_scope(principal, "mail:profile:write"): + return + if _is_own_user_scope(principal, scope_type, scope_id) and has_scope( + principal, "mail:profile:write_own" + ): + return + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Missing scope: mail:profile:write", + ) + + +def _require_profile_credentials_scope( + principal: ApiPrincipal, + scope_type: str, + scope_id: str | None = None, +) -> None: + if scope_type == "system": + _require_scope(principal, "system:settings:write") + return + if has_scope(principal, "mail:secret:manage"): + return + if _is_own_user_scope(principal, scope_type, scope_id) and has_scope( + principal, "mail:secret:manage_own" + ): + return + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Missing scope: mail:secret:manage", + ) + + +def _profile_for_mutation( + session: Session, + *, + principal: ApiPrincipal, + profile_id: str, +): + """Resolve a profile through the actor's mutation boundary. + + Mutation routes deliberately do not apply effective transport policy here: + an owner must still be able to repair or deactivate a profile after policy + becomes more restrictive. Missing and non-owned identifiers are kept + indistinguishable for self-service actors. + """ + + try: + profile = get_mail_server_profile( + session, + tenant_id=principal.tenant_id, + profile_id=profile_id, + for_update=True, + mutation_owner_user_id=( + principal.user.id + if has_scope(principal, "mail:profile:write_own") + else None + ), + can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal), + can_manage_system_profiles=_principal_can_manage_system_profiles(principal), + ) + _require_profile_write_scope( + principal, + profile.scope_type or "tenant", + profile.scope_id, + ) + except (MailProfileError, HTTPException) as exc: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Mail-server profile not found", + ) from exc + return profile + + +_TRANSPORT_ENDPOINT_FIELDS = ("host", "port", "security") + + +def _transport_endpoint_changed( + current: dict[str, Any] | None, + updated: SmtpConfig | ImapConfig, +) -> bool: + current_payload = current or {} + updated_payload = updated.model_dump(mode="json") + return any( + current_payload.get(field) != updated_payload.get(field) + for field in _TRANSPORT_ENDPOINT_FIELDS + ) + + +def _profile_update_requires_credentials_scope( + profile, + *, + payload: MailServerProfileUpdateRequest, + smtp: SmtpConfig | None, + imap: ImapConfig | None, +) -> bool: + if payload.credentials_supplied() or payload.clear_imap: + return True + if ( + smtp is not None + and getattr(profile, "smtp_password_encrypted", None) + and _transport_endpoint_changed(profile.smtp_config, smtp) + ): + return True + return bool( + imap is not None + and getattr(profile, "imap_password_encrypted", None) + and _transport_endpoint_changed(profile.imap_config, imap) + ) + + +def _profile_has_stored_credentials(profile) -> bool: + return bool( + getattr(profile, "smtp_password_encrypted", None) + or getattr(profile, "imap_password_encrypted", None) + ) def _require_mailbox_read_scope(principal: ApiPrincipal) -> None: @@ -154,6 +269,10 @@ def _principal_can_manage_tenant_profiles(principal: ApiPrincipal) -> bool: return has_scope(principal, "mail:profile:write") +def _principal_can_manage_own_profiles(principal: ApiPrincipal) -> bool: + return has_scope(principal, "mail:profile:write_own") + + def _principal_can_manage_system_profiles(principal: ApiPrincipal) -> bool: return has_scope(principal, "system:settings:write") @@ -166,6 +285,7 @@ def _profile_actor_kwargs( return { "actor_user_id": principal.user.id, "actor_group_ids": principal.group_ids, + "actor_can_manage_own_profiles": _principal_can_manage_own_profiles(principal), "actor_can_manage_tenant_profiles": _principal_can_manage_tenant_profiles(principal), "actor_can_manage_system_profiles": _principal_can_manage_system_profiles(principal), "actor_tenant_admin": _principal_is_tenant_admin(principal), @@ -187,6 +307,7 @@ def _get_profile_for_principal( profile_id=profile_id, user_id=principal.user.id, group_ids=principal.group_ids, + can_manage_own_profiles=_principal_can_manage_own_profiles(principal), can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal), can_manage_system_profiles=_principal_can_manage_system_profiles(principal), tenant_admin=_principal_is_tenant_admin(principal), @@ -444,6 +565,7 @@ def _profile_change_visible_to_principal( tenant_id=principal.tenant_id, user_id=principal.user.id, group_ids=principal.group_ids, + can_manage_own_profiles=_principal_can_manage_own_profiles(principal), can_manage_tenant_profiles=_principal_can_manage_tenant_profiles(principal), can_manage_system_profiles=_principal_can_manage_system_profiles(principal), tenant_admin=_principal_is_tenant_admin(principal), @@ -812,7 +934,12 @@ def create_profile( principal: ApiPrincipal = Depends(get_api_principal), session: Session = Depends(get_session), ): - _require_profile_write_scope(principal, payload.scope_type) + authorization_scope_id = ( + payload.scope_id or principal.user.id + if payload.scope_type == "user" + else payload.scope_id + ) + _require_profile_write_scope(principal, payload.scope_type, authorization_scope_id) _require_campaign_profile_mutation_access( session, principal=principal, @@ -820,7 +947,11 @@ def create_profile( scope_id=payload.scope_id, ) if payload.credentials_supplied(): - _require_profile_credentials_scope(principal, payload.scope_type) + _require_profile_credentials_scope( + principal, + payload.scope_type, + authorization_scope_id, + ) try: profile = create_mail_server_profile( session, @@ -881,16 +1012,17 @@ def update_profile( session: Session = Depends(get_session), ): try: - profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) - _require_profile_write_scope(principal, profile.scope_type or "tenant") + profile = _profile_for_mutation( + session, + principal=principal, + profile_id=profile_id, + ) _require_campaign_profile_mutation_access( session, principal=principal, scope_type=profile.scope_type or "tenant", scope_id=profile.scope_id, ) - if payload.credentials_supplied() or payload.clear_imap: - _require_profile_credentials_scope(principal, profile.scope_type or "tenant") smtp_config = payload.smtp_config() if payload.smtp is None and "smtp" in payload.credentials.model_fields_set: smtp_data = dict(profile.smtp_config or {}) @@ -901,6 +1033,17 @@ def update_profile( imap_data = dict(profile.imap_config or {}) imap_data.update(payload.credentials.imap.model_dump(mode="json", exclude_unset=True)) imap_config = ImapConfig.model_validate(imap_data) + if _profile_update_requires_credentials_scope( + profile, + payload=payload, + smtp=smtp_config, + imap=imap_config, + ): + _require_profile_credentials_scope( + principal, + profile.scope_type or "tenant", + profile.scope_id, + ) update_mail_server_profile( session, profile, @@ -943,15 +1086,23 @@ def deactivate_profile( session: Session = Depends(get_session), ): try: - profile = get_mail_server_profile(session, tenant_id=principal.tenant_id, profile_id=profile_id) - _require_profile_write_scope(principal, profile.scope_type or "tenant") + profile = _profile_for_mutation( + session, + principal=principal, + profile_id=profile_id, + ) _require_campaign_profile_mutation_access( session, principal=principal, scope_type=profile.scope_type or "tenant", scope_id=profile.scope_id, ) - _require_profile_credentials_scope(principal, profile.scope_type or "tenant") + if _profile_has_stored_credentials(profile): + _require_profile_credentials_scope( + principal, + profile.scope_type or "tenant", + profile.scope_id, + ) was_active = bool(profile.is_active) profile.is_active = False deleted_protocols = delete_mail_profile_credentials( diff --git a/tests/test_documentation.py b/tests/test_documentation.py index 772ea61..1d00298 100644 --- a/tests/test_documentation.py +++ b/tests/test_documentation.py @@ -70,6 +70,20 @@ class MailRuntimeDocumentationTests(unittest.TestCase): self.assertEqual(task.metadata["help_contexts"], ["mail.profiles", "app.settings"]) self.assertIn("current account's user scope", task.metadata["prerequisites"][0]) + self_service = self.topics( + self.context( + {"mail:profile:read", "mail:profile:write_own"} + ), + enabled, + ) + self.assertIn("mail.workflow.create-custom-profile", self_service) + condition = self_service["mail.workflow.create-custom-profile"].conditions[0] + self.assertEqual(condition.required_scopes, ("mail:profile:read",)) + self.assertEqual( + condition.any_scopes, + ("mail:profile:write", "mail:profile:write_own"), + ) + def test_custom_profile_task_distinguishes_secret_test_and_use_authority(self) -> None: policy = EffectiveMailProfilePolicy() cases = ( @@ -79,6 +93,7 @@ class MailRuntimeDocumentationTests(unittest.TestCase): ({"mail:profile:use"}, False, False, True), ({"mail:profile:test", "mail:profile:use"}, False, True, True), ({"mail:secret:manage", "mail:profile:test", "mail:profile:use"}, True, True, True), + ({"mail:secret:manage_own"}, True, False, False), ) for extra_scopes, credentials, testing, use in cases: with self.subTest(extra_scopes=extra_scopes): diff --git a/tests/test_manifest.py b/tests/test_manifest.py index ba8026d..ee05c5f 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -40,6 +40,21 @@ class MailManifestTests(unittest.TestCase): for interface in manifest.provides_interfaces ], ) + permissions = {permission.scope for permission in manifest.permissions} + self.assertIn("mail:profile:write_own", permissions) + self.assertIn("mail:secret:manage_own", permissions) + roles = {template.slug: template for template in manifest.role_templates} + self.assertEqual( + set(roles["mail_profile_self_service"].permissions), + { + "mail:profile:read", + "mail:profile:use", + "mail:profile:test", + "mail:mailbox:read", + "mail:profile:write_own", + "mail:secret:manage_own", + }, + ) topics = {topic.id: topic for topic in manifest.documentation} self.assertTrue( { diff --git a/tests/test_profile_actor_authorization.py b/tests/test_profile_actor_authorization.py index c192635..e6d60ca 100644 --- a/tests/test_profile_actor_authorization.py +++ b/tests/test_profile_actor_authorization.py @@ -2,14 +2,17 @@ from __future__ import annotations import unittest from types import SimpleNamespace -from unittest.mock import patch +from unittest.mock import ANY, patch from fastapi import HTTPException +from sqlalchemy.dialects import postgresql from govoplan_mail.backend import router +from govoplan_mail.backend import mail_profiles from govoplan_mail.backend.mail_profiles import ( EffectiveMailProfilePolicy, MailProfileError, + get_mail_server_profile_for_actor, list_mail_server_profiles, mail_profile_visible_to_actor, ) @@ -35,6 +38,8 @@ def _profile( name=profile_id, smtp_config={"host": "smtp.example.test"}, imap_config={"host": "imap.example.test"}, + smtp_password_encrypted=None, + imap_password_encrypted=None, ) @@ -67,11 +72,15 @@ class _Session: def rollback(self): self.rollbacks += 1 + def refresh(self, _value): + return None + class _Principal: tenant_id = "tenant-1" user = SimpleNamespace(id="user-1") group_ids = frozenset({"group-1"}) + api_key_id = None def __init__(self, scopes): self._scopes = frozenset(scopes) @@ -81,6 +90,373 @@ class _Principal: class ProfileActorAuthorizationTests(unittest.TestCase): + def test_profile_mutation_selector_is_owner_bounded_and_row_locked(self) -> None: + statement = mail_profiles._mail_server_profile_mutation_statement( # noqa: SLF001 + tenant_id="tenant-1", + profile_id="profile-1", + owner_user_id="user-1", + can_manage_tenant_profiles=False, + can_manage_system_profiles=False, + ) + + sql = str(statement.compile(dialect=postgresql.dialect())) + self.assertIn("FOR UPDATE", sql) + self.assertIn("mail_server_profiles.tenant_id =", sql) + self.assertIn("mail_server_profiles.scope_type =", sql) + self.assertIn("mail_server_profiles.scope_id =", sql) + self.assertTrue(statement.get_execution_options()["populate_existing"]) + + def test_profile_mutation_selector_without_authority_cannot_lock_a_row(self) -> None: + statement = mail_profiles._mail_server_profile_mutation_statement( # noqa: SLF001 + tenant_id="tenant-1", + profile_id="profile-1", + owner_user_id=None, + can_manage_tenant_profiles=False, + can_manage_system_profiles=False, + ) + + sql = str(statement.compile(dialect=postgresql.dialect())) + self.assertIn("mail_server_profiles.id IS NULL", sql) + self.assertIn("FOR UPDATE", sql) + + def test_self_service_profile_permissions_are_limited_to_own_user_scope(self) -> None: + principal = _Principal( + {"mail:profile:write_own", "mail:secret:manage_own"} + ) + + router._require_profile_write_scope( # noqa: SLF001 - authorization seam + principal, # type: ignore[arg-type] + "user", + "user-1", + ) + router._require_profile_credentials_scope( # noqa: SLF001 - authorization seam + principal, # type: ignore[arg-type] + "user", + "user-1", + ) + + for scope_type, scope_id in ( + ("user", "user-2"), + ("tenant", "tenant-1"), + ("group", "group-1"), + ("campaign", "campaign-1"), + ("system", None), + ): + with self.subTest(scope_type=scope_type, scope_id=scope_id): + with self.assertRaises(HTTPException) as write_denied: + router._require_profile_write_scope( # noqa: SLF001 + principal, # type: ignore[arg-type] + scope_type, + scope_id, + ) + self.assertEqual(write_denied.exception.status_code, 403) + + with self.assertRaises(HTTPException) as secret_denied: + router._require_profile_credentials_scope( # noqa: SLF001 + principal, # type: ignore[arg-type] + scope_type, + scope_id, + ) + self.assertEqual(secret_denied.exception.status_code, 403) + + def test_self_service_create_rejects_another_user_before_persistence(self) -> None: + principal = _Principal( + { + "mail:profile:read", + "mail:profile:write_own", + "mail:secret:manage_own", + } + ) + payload = MailServerProfileCreateRequest.model_validate( + { + "name": "Other user's profile", + "scope_type": "user", + "scope_id": "user-2", + "smtp": { + "host": "smtp.example.test", + "password": "not-persisted", + }, + } + ) + with ( + patch("govoplan_mail.backend.router.create_mail_server_profile") as create, + self.assertRaises(HTTPException) as denied, + ): + router.create_profile( + payload, + principal=principal, # type: ignore[arg-type] + session=_Session(), # type: ignore[arg-type] + ) + + self.assertEqual(denied.exception.status_code, 403) + create.assert_not_called() + + def test_self_service_update_hides_another_user_before_mutation(self) -> None: + principal = _Principal( + {"mail:profile:write_own", "mail:secret:manage_own"} + ) + profile = _profile( + "other-user-profile", + scope_type="user", + scope_id="user-2", + ) + with ( + patch( + "govoplan_mail.backend.router.get_mail_server_profile", + return_value=profile, + ), + patch("govoplan_mail.backend.router.update_mail_server_profile") as update, + self.assertRaises(HTTPException) as denied, + ): + router.update_profile( + profile.id, + MailServerProfileUpdateRequest(name="Must not change"), + principal=principal, # type: ignore[arg-type] + session=_Session(), # type: ignore[arg-type] + ) + + self.assertEqual(denied.exception.status_code, 404) + update.assert_not_called() + + def test_self_service_delete_hides_another_user_before_mutation(self) -> None: + principal = _Principal( + {"mail:profile:write_own", "mail:secret:manage_own"} + ) + profile = _profile( + "other-user-profile", + scope_type="user", + scope_id="user-2", + ) + with ( + patch( + "govoplan_mail.backend.router.get_mail_server_profile", + return_value=profile, + ), + patch("govoplan_mail.backend.router.delete_mail_profile_credentials") as delete, + self.assertRaises(HTTPException) as denied, + ): + router.deactivate_profile( + profile.id, + principal=principal, # type: ignore[arg-type] + session=_Session(), # type: ignore[arg-type] + ) + + self.assertEqual(denied.exception.status_code, 404) + delete.assert_not_called() + + def test_mutation_lookup_returns_the_same_not_found_for_missing_and_non_owned(self) -> None: + principal = _Principal({"mail:profile:write_own"}) + other_profile = _profile( + "other-user-profile", + scope_type="user", + scope_id="user-2", + ) + cases = ( + (other_profile, None), + (None, MailProfileError("Mail-server profile not found")), + ) + for returned, failure in cases: + with self.subTest(failure=failure is not None): + kwargs = {"side_effect": failure} if failure else {"return_value": returned} + with ( + patch( + "govoplan_mail.backend.router.get_mail_server_profile", + **kwargs, + ), + self.assertRaises(HTTPException) as denied, + ): + router._profile_for_mutation( # noqa: SLF001 - authorization seam + _Session(), # type: ignore[arg-type] + principal=principal, # type: ignore[arg-type] + profile_id="candidate-id", + ) + + self.assertEqual(denied.exception.status_code, 404) + self.assertEqual(denied.exception.detail, "Mail-server profile not found") + + def test_broad_profile_admin_retains_cross_owner_mutation_access(self) -> None: + principal = _Principal({"mail:profile:write"}) + profile = _profile( + "other-user-profile", + scope_type="user", + scope_id="user-2", + ) + with patch( + "govoplan_mail.backend.router.get_mail_server_profile", + return_value=profile, + ) as get_profile: + resolved = router._profile_for_mutation( # noqa: SLF001 - authorization seam + _Session(), # type: ignore[arg-type] + principal=principal, # type: ignore[arg-type] + profile_id=profile.id, + ) + + self.assertIs(resolved, profile) + get_profile.assert_called_once_with( + ANY, + tenant_id="tenant-1", + profile_id=profile.id, + for_update=True, + mutation_owner_user_id=None, + can_manage_tenant_profiles=True, + can_manage_system_profiles=False, + ) + + def test_self_service_transport_rebind_requires_own_secret_authority(self) -> None: + principal = _Principal({"mail:profile:write_own"}) + profile = _profile( + "own-user-profile", + scope_type="user", + scope_id="user-1", + ) + profile.smtp_config = { + "host": "smtp.example.test", + "port": 587, + "security": "starttls", + "timeout_seconds": 30, + } + profile.smtp_password_encrypted = "existing-ciphertext" + payload = MailServerProfileUpdateRequest.model_validate( + {"smtp": {"host": "smtp.attacker.test"}} + ) + session = _Session() + with ( + patch( + "govoplan_mail.backend.router.get_mail_server_profile", + return_value=profile, + ), + patch("govoplan_mail.backend.router.update_mail_server_profile") as update, + self.assertRaises(HTTPException) as denied, + ): + router.update_profile( + profile.id, + payload, + principal=principal, # type: ignore[arg-type] + session=session, # type: ignore[arg-type] + ) + + self.assertEqual(denied.exception.status_code, 403) + self.assertEqual(session.rollbacks, 1) + update.assert_not_called() + + def test_self_service_imap_rebind_requires_own_secret_authority(self) -> None: + principal = _Principal({"mail:profile:write_own"}) + profile = _profile( + "own-user-profile", + scope_type="user", + scope_id="user-1", + ) + profile.imap_config = { + "host": "imap.example.test", + "port": 993, + "security": "tls", + "sent_folder": "auto", + "timeout_seconds": 30, + } + profile.imap_password_encrypted = "existing-ciphertext" + payload = MailServerProfileUpdateRequest.model_validate( + {"imap": {"host": "imap.attacker.test"}} + ) + with ( + patch( + "govoplan_mail.backend.router.get_mail_server_profile", + return_value=profile, + ), + patch("govoplan_mail.backend.router.update_mail_server_profile") as update, + self.assertRaises(HTTPException) as denied, + ): + router.update_profile( + profile.id, + payload, + principal=principal, # type: ignore[arg-type] + session=_Session(), # type: ignore[arg-type] + ) + + self.assertEqual(denied.exception.status_code, 403) + update.assert_not_called() + + def test_self_service_transport_rebind_is_allowed_with_own_secret_authority(self) -> None: + principal = _Principal( + {"mail:profile:write_own", "mail:secret:manage_own"} + ) + profile = _profile( + "own-user-profile", + scope_type="user", + scope_id="user-1", + ) + profile.smtp_config = { + "host": "smtp.example.test", + "port": 587, + "security": "starttls", + "timeout_seconds": 30, + } + profile.smtp_password_encrypted = "existing-ciphertext" + payload = MailServerProfileUpdateRequest.model_validate( + {"smtp": {"host": "smtp.allowed.test"}} + ) + session = _Session() + with ( + patch( + "govoplan_mail.backend.router.get_mail_server_profile", + return_value=profile, + ), + patch( + "govoplan_mail.backend.router.update_mail_server_profile", + return_value=profile, + ) as update, + patch("govoplan_mail.backend.router._record_mail_change"), + patch( + "govoplan_mail.backend.router._profile_response", + return_value=profile, + ), + ): + result = router.update_profile( + profile.id, + payload, + principal=principal, # type: ignore[arg-type] + session=session, # type: ignore[arg-type] + ) + + self.assertIs(result, profile) + self.assertEqual(session.commits, 1) + update.assert_called_once() + + def test_transport_endpoint_change_without_stored_secret_needs_only_write_authority(self) -> None: + principal = _Principal({"mail:profile:write_own"}) + profile = _profile( + "own-user-profile", + scope_type="user", + scope_id="user-1", + ) + payload = MailServerProfileUpdateRequest.model_validate( + {"smtp": {"host": "smtp.allowed.test"}} + ) + session = _Session() + with ( + patch( + "govoplan_mail.backend.router.get_mail_server_profile", + return_value=profile, + ), + patch( + "govoplan_mail.backend.router.update_mail_server_profile", + return_value=profile, + ) as update, + patch("govoplan_mail.backend.router._record_mail_change"), + patch( + "govoplan_mail.backend.router._profile_response", + return_value=profile, + ), + ): + router.update_profile( + profile.id, + payload, + principal=principal, # type: ignore[arg-type] + session=session, # type: ignore[arg-type] + ) + + self.assertEqual(session.commits, 1) + update.assert_called_once() + def test_general_profile_list_hides_other_owner_contexts(self) -> None: profiles = [ _profile("system", scope_type="system", scope_id=None, tenant_id=None), @@ -142,6 +518,63 @@ class ProfileActorAuthorizationTests(unittest.TestCase): self.assertEqual([item.id for item in visible], ["inactive"]) + def test_self_service_repair_visibility_is_limited_to_the_exact_owner(self) -> None: + own_profile = _profile( + "own-policy-invalid", + scope_type="user", + scope_id="user-1", + ) + other_profile = _profile( + "other-policy-invalid", + scope_type="user", + scope_id="user-2", + ) + denied = EffectiveMailProfilePolicy( + allowed_profile_id_sets=[{"different-profile"}] + ) + session = _Session([own_profile, other_profile]) + + with patch( + "govoplan_mail.backend.mail_profiles.effective_mail_profile_policy", + return_value=denied, + ): + ordinary_visible = list_mail_server_profiles( + session, # type: ignore[arg-type] + tenant_id="tenant-1", + include_inactive=True, + actor_user_id="user-1", + actor_administrative_visibility=True, + ) + repair_visible = list_mail_server_profiles( + session, # type: ignore[arg-type] + tenant_id="tenant-1", + include_inactive=True, + actor_user_id="user-1", + actor_can_manage_own_profiles=True, + actor_administrative_visibility=True, + ) + resolved = get_mail_server_profile_for_actor( + session, # type: ignore[arg-type] + tenant_id="tenant-1", + profile_id=own_profile.id, + user_id="user-1", + can_manage_own_profiles=True, + administrative_visibility=True, + ) + with self.assertRaises(MailProfileError): + get_mail_server_profile_for_actor( + session, # type: ignore[arg-type] + tenant_id="tenant-1", + profile_id=other_profile.id, + user_id="user-1", + can_manage_own_profiles=True, + administrative_visibility=True, + ) + + self.assertEqual(ordinary_visible, []) + self.assertEqual([item.id for item in repair_visible], [own_profile.id]) + self.assertIs(resolved, own_profile) + def test_profile_admin_can_list_but_not_use_a_policy_denied_profile(self) -> None: profile = _profile("denied", scope_type="tenant", scope_id="tenant-1") session = _Session([profile]) diff --git a/tests/test_router_profile_deletion.py b/tests/test_router_profile_deletion.py index 96b7c5d..cb7a19f 100644 --- a/tests/test_router_profile_deletion.py +++ b/tests/test_router_profile_deletion.py @@ -34,6 +34,7 @@ class MailProfileDeletionRouteTests(unittest.TestCase): tenant_id="tenant-1", user=SimpleNamespace(id="user-1"), api_key_id=None, + has=lambda _scope: False, ) self.profile = SimpleNamespace( id="profile-1", @@ -48,7 +49,7 @@ class MailProfileDeletionRouteTests(unittest.TestCase): with ( patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile), patch("govoplan_mail.backend.router._require_profile_write_scope"), - patch("govoplan_mail.backend.router._require_profile_credentials_scope"), + patch("govoplan_mail.backend.router._require_profile_credentials_scope") as require_credentials, patch("govoplan_mail.backend.router.delete_mail_profile_credentials", return_value=()), patch("govoplan_mail.backend.router.clear_mailbox_index") as clear_index, patch("govoplan_mail.backend.router._record_mail_change") as record_change, @@ -61,6 +62,32 @@ class MailProfileDeletionRouteTests(unittest.TestCase): self.assertEqual(session.rollbacks, 0) clear_index.assert_called_once_with(session, profile_id="profile-1") record_change.assert_not_called() + require_credentials.assert_not_called() + + def test_delete_requires_secret_authority_when_credentials_will_be_deleted(self) -> None: + self.profile.is_active = True + self.profile.smtp_password_encrypted = "existing-ciphertext" + self.profile.imap_password_encrypted = None + session = _Session() + with ( + patch("govoplan_mail.backend.router.get_mail_server_profile", return_value=self.profile), + patch("govoplan_mail.backend.router._require_profile_write_scope"), + patch("govoplan_mail.backend.router._require_profile_credentials_scope") as require_credentials, + patch( + "govoplan_mail.backend.router.delete_mail_profile_credentials", + return_value=("smtp",), + ), + patch("govoplan_mail.backend.router.clear_mailbox_index"), + patch("govoplan_mail.backend.router._record_mail_change"), + patch("govoplan_mail.backend.router._profile_response", return_value=self.profile), + ): + deactivate_profile("profile-1", principal=self.principal, session=session) + + require_credentials.assert_called_once_with( + self.principal, + "tenant", + "tenant-1", + ) def test_change_feed_reports_whether_credentials_were_deleted(self) -> None: self.profile.is_active = True