5 Commits

17 changed files with 1415 additions and 59 deletions

View File

@@ -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:

View File

@@ -114,6 +114,10 @@ non-production provider and mailbox first. A successful connection test does
not prove policy authorization for a later Campaign context, deliverability,
recipient acceptance, SPF/DKIM/DMARC alignment, or future availability.
Testing a saved profile requires both `mail:profile:test` and
`mail:profile:use`, and the profile must be active. Profile creation or test
authority alone is not enough.
Raw settings test endpoints accept new settings only for actors who may both
test profiles and manage secrets. They are an administration aid, not a way for
ordinary consumers to bypass reusable profiles.
@@ -138,17 +142,39 @@ 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 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. 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
behavior, and timeouts. Sender/envelope/recipient constraints belong to

View File

@@ -1,6 +1,6 @@
{
"name": "@govoplan/mail-webui",
"version": "0.1.9",
"version": "0.1.10",
"private": true,
"type": "module",
"main": "webui/src/index.ts",
@@ -19,7 +19,7 @@
"LICENSE"
],
"peerDependencies": {
"@govoplan/core-webui": "^0.1.9",
"@govoplan/core-webui": "^0.1.10",
"lucide-react": "^1.23.0",
"react": "^19.0.0",
"react-dom": "^19.0.0",

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "govoplan-mail"
version = "0.1.9"
version = "0.1.10"
description = "GovOPlaN mail module with backend and WebUI integration."
readme = "README.md"
requires-python = ">=3.12"

View File

@@ -4,15 +4,35 @@ from typing import Any
from sqlalchemy.orm import Session
from govoplan_core.core.modules import DocumentationContext, DocumentationLink, DocumentationTopic
from govoplan_mail.backend.mail_profiles import MailProfileError, effective_mail_profile_policy_for_scope
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"
# RBAC permission identifiers; neither value is a stored credential.
MAIL_SECRET_MANAGE_SCOPE = "mail:secret:manage" # noqa: S105 # nosec B105
MAIL_SECRET_MANAGE_OWN_SCOPE = "mail:secret:manage_own" # noqa: S105 # nosec B105
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, ...]:
topic = _tenant_mail_policy_topic(context)
return (topic,) if topic is not None else ()
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:
@@ -29,17 +49,26 @@ def _tenant_mail_policy_topic(context: DocumentationContext) -> DocumentationTop
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=str(exc),
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=(_mail_policy_api_link(),),
metadata={"error_type": type(exc).__name__},
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()
@@ -90,6 +119,243 @@ def _tenant_mail_policy_topic(context: DocumentationContext) -> DocumentationTop
)
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)
@@ -259,5 +525,16 @@ def _has_any_scope(principal: object | None, scopes: tuple[str, ...]) -> bool:
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")

View File

@@ -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,7 +1219,21 @@ 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:
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")
@@ -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

View File

@@ -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=(
@@ -279,6 +307,7 @@ manifest = ModuleManifest(
"kind": "workflow",
"route": "/settings?section=mail-profiles",
"screen": "Mail profiles",
"help_contexts": ["mail.profiles", "app.settings"],
"prerequisites": [
"Mail is installed and you may read, use, and test profiles visible in the current context.",
"A profile administrator has configured credentials and effective policy.",
@@ -321,6 +350,7 @@ manifest = ModuleManifest(
"kind": "workflow",
"route": "/mail",
"screen": "Mail",
"help_contexts": ["mail.list"],
"prerequisites": [
"An active visible profile has IMAP configured.",
"You may both use that profile and read its mailbox.",

View File

@@ -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(

186
tests/test_documentation.py Normal file
View File

@@ -0,0 +1,186 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from unittest.mock import patch
from sqlalchemy.orm import Session
from govoplan_core.core.modules import DocumentationContext
from govoplan_mail.backend.documentation import documentation_topics
from govoplan_mail.backend.mail_profiles import EffectiveMailProfilePolicy, MailProfileError
class _Principal:
tenant_id = "tenant-1"
user = SimpleNamespace(id="user-1")
def __init__(self, scopes: set[str]) -> None:
self.scopes = frozenset(scopes)
def has(self, scope: str) -> bool:
return scope in self.scopes
class MailRuntimeDocumentationTests(unittest.TestCase):
def setUp(self) -> None:
self.session = Session()
def tearDown(self) -> None:
self.session.close()
def context(self, scopes: set[str], *, documentation_type: str = "user") -> DocumentationContext:
return DocumentationContext(
registry=object(),
principal=_Principal(scopes),
settings=None,
session=self.session,
documentation_type=documentation_type, # type: ignore[arg-type]
)
def topics(self, context: DocumentationContext, user_policy: EffectiveMailProfilePolicy) -> dict[str, object]:
tenant_policy = EffectiveMailProfilePolicy()
def effective_policy(_session, *, scope_type, **_kwargs):
return user_policy if scope_type == "user" else tenant_policy
with patch(
"govoplan_mail.backend.documentation.effective_mail_profile_policy_for_scope",
side_effect=effective_policy,
):
return {topic.id: topic for topic in documentation_topics(context)}
def test_custom_profile_task_requires_write_authority_and_enabled_user_profiles(self) -> None:
enabled = EffectiveMailProfilePolicy(allow_user_profiles=True)
without_write = self.topics(self.context({"mail:profile:read"}), enabled)
self.assertNotIn("mail.workflow.create-custom-profile", without_write)
without_read = self.topics(self.context({"mail:profile:write"}), enabled)
self.assertNotIn("mail.workflow.create-custom-profile", without_read)
disabled = EffectiveMailProfilePolicy(allow_user_profiles=False)
blocked = self.topics(self.context({"mail:profile:read", "mail:profile:write"}), disabled)
self.assertNotIn("mail.workflow.create-custom-profile", blocked)
available = self.topics(self.context({"mail:profile:read", "mail:profile:write"}), enabled)
self.assertIn("mail.workflow.create-custom-profile", available)
task = available["mail.workflow.create-custom-profile"]
self.assertEqual(task.title, "Create a custom Mail profile")
self.assertEqual(task.metadata["route"], "/settings?section=mail-profiles")
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 = (
(set(), False, False, False),
({"mail:secret:manage"}, True, False, False),
({"mail:profile:test"}, False, False, False),
({"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):
topics = self.topics(self.context({"mail:profile:read", "mail:profile:write", *extra_scopes}), policy)
task = topics["mail.workflow.create-custom-profile"]
self.assertEqual(task.metadata["can_manage_credentials"], credentials)
self.assertEqual(task.metadata["can_test_profile"], testing)
self.assertEqual(task.metadata["can_use_profile"], use)
self.assertIn(
"you may save or replace" if credentials else "you cannot save or replace passwords",
task.body,
)
self.assertIn(
"you may run" if testing else "does not let you run connection tests",
task.body,
)
self.assertIn(
"you may select" if use else "does not let you select or use it",
task.body,
)
def test_custom_profile_task_preserves_host_allow_group_and_deny_precedence(self) -> None:
policy = EffectiveMailProfilePolicy(
whitelist_groups={
"smtp_hosts": [
["*.example.edu", "smtp.shared.test"],
["smtp-*.example.edu"],
],
"imap_hosts": [["imap.example.edu"]],
},
blacklist_patterns={
"smtp_hosts": ["smtp-blocked.example.edu"],
"imap_hosts": ["imap-legacy.example.edu"],
},
allowed_profile_id_sets=[{"private-profile-id"}],
)
topics = self.topics(
self.context({"mail:profile:read", "mail:profile:write", "mail:secret:manage", "mail:profile:test", "mail:profile:use"}),
policy,
)
task = topics["mail.workflow.create-custom-profile"]
constraints = {item["id"]: item for item in task.metadata["constraints"]}
constraints_text = " ".join(item["description"] for item in task.metadata["constraints"])
self.assertIn("Deny rules are checked first", constraints_text)
self.assertEqual(constraints["smtp-host-deny"]["values"], ["smtp-blocked.example.edu"])
self.assertEqual(constraints["smtp-host-allow-1"]["values"], ["*.example.edu", "smtp.shared.test"])
self.assertEqual(constraints["smtp-host-allow-2"]["values"], ["smtp-*.example.edu"])
self.assertIn("every active allow-list group", constraints_text)
self.assertEqual(constraints["imap-host-allow-1"]["values"], ["imap.example.edu"])
self.assertTrue(task.metadata["approval_required_before_use"])
self.assertIn("must be approved before the profile can be selected or used", task.body)
self.assertNotIn("private-profile-id", repr(task))
self.assertIn("every active allow-list group", constraints["smtp-host-allow-1"]["description"])
def test_custom_profile_task_explains_when_no_approval_list_is_active(self) -> None:
topics = self.topics(
self.context({"mail:profile:read", "mail:profile:write", "mail:profile:use"}),
EffectiveMailProfilePolicy(),
)
task = topics["mail.workflow.create-custom-profile"]
self.assertFalse(task.metadata["approval_required_before_use"])
self.assertIn("no approved-profile list currently blocks", task.body)
def test_user_policy_errors_are_generic_and_never_create_a_task(self) -> None:
context = self.context({"mail:profile:read", "mail:profile:write"})
with patch(
"govoplan_mail.backend.documentation.effective_mail_profile_policy_for_scope",
side_effect=MailProfileError("sensitive source-id profile-id username secret"),
):
topics = {topic.id: topic for topic in documentation_topics(context)}
self.assertNotIn("mail.workflow.create-custom-profile", topics)
unavailable = topics["mail.tenant-profile-policy-unavailable"]
self.assertEqual(
unavailable.body,
"The current Mail policy could not be loaded. Try again or ask a Mail administrator for help.",
)
self.assertNotIn("sensitive", repr(unavailable))
def test_manifest_workflows_declare_contextual_help_targets(self) -> None:
from govoplan_mail.backend.manifest import get_manifest
topics = {topic.id: topic for topic in get_manifest().documentation}
self.assertEqual(topics["mail.workflow.choose-and-test-profile"].metadata["help_contexts"], ["mail.profiles", "app.settings"])
self.assertEqual(topics["mail.workflow.read-mailbox"].metadata["help_contexts"], ["mail.list"])
if __name__ == "__main__":
unittest.main()

View File

@@ -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(
{

View File

@@ -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])

View File

@@ -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

View File

@@ -1,17 +1,17 @@
{
"name": "@govoplan/mail-webui",
"version": "0.1.9",
"version": "0.1.10",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "@govoplan/mail-webui",
"version": "0.1.9",
"version": "0.1.10",
"devDependencies": {
"typescript": "^5.7.2"
},
"peerDependencies": {
"@govoplan/core-webui": "^0.1.9",
"@govoplan/core-webui": "^0.1.10",
"lucide-react": "^1.23.0",
"react": "^19.0.0",
"react-dom": "^19.0.0",

View File

@@ -1,6 +1,6 @@
{
"name": "@govoplan/mail-webui",
"version": "0.1.9",
"version": "0.1.10",
"private": true,
"type": "module",
"main": "src/index.ts",
@@ -14,7 +14,7 @@
"./styles/mail-profiles.css": "./src/styles/mail-profiles.css"
},
"peerDependencies": {
"@govoplan/core-webui": "^0.1.9",
"@govoplan/core-webui": "^0.1.10",
"lucide-react": "^1.23.0",
"react": "^19.0.0",
"react-dom": "^19.0.0",

View File

@@ -42,6 +42,8 @@ import {
mailProfileEditTargetShowsProfileFields,
mailProfileEditTargetShowsSettingsPanel,
mailProfileEditTargetVisibleSections,
mailProfileCreateCredentialsPayload,
mailProfileTargetedUpdatePayload,
type MailProfileEditTarget,
type MailProfileProtocol
} from "./mailProfileEditorModel";
@@ -278,7 +280,7 @@ export function MailProfileScopeManager({
const created = await createMailServerProfile(settings, payload);
setSuccess(i18nMessage("i18n:govoplan-mail.profile_value_created.2a088d8d", { value0: created.name }));
} else {
const payload = updateProfilePayload(draft, editing);
const payload = updateProfilePayload(draft, editing, editingTarget);
const updated = await updateMailServerProfile(settings, editing.id, payload);
setSuccess(i18nMessage("i18n:govoplan-mail.profile_value_updated.fdbad0ea", { value0: updated.name }));
}
@@ -369,13 +371,16 @@ export function MailProfileScopeManager({
}
if (row.kind === "credential") {
return <TableActionGroup actions={[{
id: "edit-credentials", label: i18nMessage("i18n:govoplan-mail.edit_value_credentials.1e2dc4f6", { value0: row.protocol.toUpperCase() }), icon: <Pencil size={16} />, disabled: !canWriteProfiles || busy, onClick: () => openEdit(row.profile, { kind: "credentials", protocol: row.protocol })
id: "edit-credentials", label: i18nMessage("i18n:govoplan-mail.edit_value_credentials.1e2dc4f6", { value0: row.protocol.toUpperCase() }), icon: <Pencil size={16} />, disabled: !canWriteProfiles || !canManageCredentials || busy, onClick: () => openEdit(row.profile, { kind: "credentials", protocol: row.protocol })
}]} />;
}
const deactivationDeletesCredentials = Boolean(
row.profile.smtp_password_configured || row.profile.imap_password_configured
);
return <TableActionGroup actions={[
{ id: "edit", label: i18nMessage("i18n:govoplan-mail.edit_value.fad75899", { value0: row.profile.name }), icon: <Pencil size={16} />, disabled: !canWriteProfiles || busy, onClick: () => openEdit(row.profile) },
{ id: "deactivate", label: i18nMessage("i18n:govoplan-mail.deactivate_value.a276a667", { value0: row.profile.name }), icon: <Trash2 size={16} />, variant: "danger", applicable: row.profile.is_active, disabled: !canWriteProfiles || busy, onClick: () => setDeactivating(row.profile) }
{ id: "deactivate", label: i18nMessage("i18n:govoplan-mail.deactivate_value.a276a667", { value0: row.profile.name }), icon: <Trash2 size={16} />, variant: "danger", applicable: row.profile.is_active, disabled: !canWriteProfiles || busy || (deactivationDeletesCredentials && !canManageCredentials), onClick: () => setDeactivating(row.profile) }
]} />;
}
@@ -1092,6 +1097,7 @@ function profileToDraft(profile: MailServerProfile): ProfileDraft {
}
function createProfilePayload(draft: ProfileDraft, scopeType: MailProfileScope, scopeId: string | null): MailServerProfilePayload {
const credentials = profileCreateCredentialsPayload(draft);
return {
name: draft.name.trim(),
slug: mailTextOrNull(draft.slug),
@@ -1101,21 +1107,27 @@ function createProfilePayload(draft: ProfileDraft, scopeType: MailProfileScope,
scope_id: scopeId,
smtp: smtpServerPayload(draft),
imap: hasDraftImapSettings(draft) ? imapServerPayload(draft) : null,
credentials: profileCredentialsPayload(draft, false)
...(credentials ? { credentials } : {})
};
}
function updateProfilePayload(draft: ProfileDraft, profile: MailServerProfile): MailServerProfileUpdatePayload {
return {
function updateProfilePayload(
draft: ProfileDraft,
profile: MailServerProfile,
target: MailProfileEditTarget
): MailServerProfileUpdatePayload {
return mailProfileTargetedUpdatePayload(target, {
profile: {
name: draft.name.trim(),
slug: mailTextOrNull(draft.slug),
description: draft.description.trim(),
is_active: draft.isActive,
is_active: draft.isActive
},
smtp: smtpServerPayload(draft),
imap: hasDraftImapSettings(draft) ? imapServerPayload(draft) : null,
credentials: profileCredentialsPayload(draft, true),
clear_imap: !hasDraftImapSettings(draft) && Boolean(profile.imap)
};
clearImap: !hasDraftImapSettings(draft) && Boolean(profile.imap)
}) as MailServerProfileUpdatePayload;
}
function smtpServerPayload(draft: ProfileDraft): MailSmtpTestPayload {
@@ -1139,6 +1151,13 @@ function profileCredentialsPayload(draft: ProfileDraft, preserveBlankPassword: b
};
}
function profileCreateCredentialsPayload(draft: ProfileDraft) {
return mailProfileCreateCredentialsPayload(
mailTransportCredentialsPayload(draft.smtpUsername, draft.smtpPassword, true),
mailTransportCredentialsPayload(draft.imapUsername, draft.imapPassword, true)
);
}
function rawSmtpPayload(draft: ProfileDraft, preserveBlankPassword: boolean): MailSmtpTestPayload {
return { ...smtpServerPayload(draft), ...mailTransportCredentialsPayload(draft.smtpUsername, draft.smtpPassword, preserveBlankPassword) };
}

View File

@@ -23,6 +23,22 @@ export type MailProfileChildDescriptor = {
protocol: MailProfileProtocol;
};
export type MailProfileTransportCredentialsLike = {
username?: string | null;
password?: string | null;
};
export type MailProfileTargetedUpdateParts = {
profile: Record<string, unknown>;
smtp: Record<string, unknown>;
imap: Record<string, unknown> | null;
credentials: {
smtp: MailProfileTransportCredentialsLike;
imap: MailProfileTransportCredentialsLike;
};
clearImap: boolean;
};
export function mailProfileChildDescriptors(profile: MailProfileTreeProfileLike): MailProfileChildDescriptor[] {
const children: MailProfileChildDescriptor[] = [
{ kind: "server", id: `server:${profile.id}:smtp`, protocol: "smtp" },
@@ -58,3 +74,42 @@ export function mailProfileEditTargetShowsProfileFields(target: MailProfileEditT
export function mailProfileEditTargetShowsSettingsPanel(target: MailProfileEditTarget): boolean {
return target.kind !== "profile";
}
/**
* Keep the PATCH authority surface equal to the focused editor. Profile and
* server edits must not accidentally replay credential fields merely because
* the form draft contains their displayed values.
*/
export function mailProfileTargetedUpdatePayload(
target: MailProfileEditTarget,
parts: MailProfileTargetedUpdateParts
): Record<string, unknown> {
if (target.kind === "profile") return parts.profile;
if (target.kind === "server") {
if (target.protocol === "smtp") return { smtp: parts.smtp };
return parts.imap === null
? { imap: null, clear_imap: parts.clearImap }
: { imap: parts.imap };
}
if (target.kind === "credentials") {
return { credentials: { [target.protocol]: parts.credentials[target.protocol] } };
}
throw new Error("Create is not an update target");
}
/** Omit blank create credentials so profile-write remains independent. */
export function mailProfileCreateCredentialsPayload(
smtp: MailProfileTransportCredentialsLike,
imap: MailProfileTransportCredentialsLike
): {smtp?: MailProfileTransportCredentialsLike;imap?: MailProfileTransportCredentialsLike;} | undefined {
const populated = (value: MailProfileTransportCredentialsLike): MailProfileTransportCredentialsLike =>
Object.fromEntries(
Object.entries(value).filter(([, item]) => item !== null && item !== undefined && item !== "")
);
const smtpCredentials = populated(smtp);
const imapCredentials = populated(imap);
const result: {smtp?: MailProfileTransportCredentialsLike;imap?: MailProfileTransportCredentialsLike;} = {};
if (Object.keys(smtpCredentials).length > 0) result.smtp = smtpCredentials;
if (Object.keys(imapCredentials).length > 0) result.imap = imapCredentials;
return Object.keys(result).length > 0 ? result : undefined;
}

View File

@@ -18,7 +18,9 @@ import {
mailProfileEditTargetPanelMode,
mailProfileEditTargetShowsProfileFields,
mailProfileEditTargetShowsSettingsPanel,
mailProfileEditTargetVisibleSections
mailProfileEditTargetVisibleSections,
mailProfileCreateCredentialsPayload,
mailProfileTargetedUpdatePayload
} from "../src/features/mail/mailProfileEditorModel";
const smtpOnlyChildren = mailProfileChildDescriptors({ id: "profile-1", imap: null });
@@ -46,3 +48,42 @@ assertEqual(mailProfileEditTargetShowsProfileFields({ kind: "profile" }), true);
assertEqual(mailProfileEditTargetShowsProfileFields({ kind: "server", protocol: "smtp" }), false);
assertEqual(mailProfileEditTargetShowsSettingsPanel({ kind: "profile" }), false);
assertEqual(mailProfileEditTargetShowsSettingsPanel({ kind: "create" }), true);
const updateParts = {
profile: { name: "Renamed" },
smtp: { host: "smtp.example.org" },
imap: { host: "imap.example.org" },
credentials: {
smtp: { username: "smtp-user", password: "smtp-secret" },
imap: { username: "imap-user", password: "imap-secret" }
},
clearImap: false
};
assertDeepEqual(
mailProfileTargetedUpdatePayload({ kind: "profile" }, updateParts),
{ name: "Renamed" },
"profile edits do not replay transport or credential fields"
);
assertDeepEqual(
mailProfileTargetedUpdatePayload({ kind: "server", protocol: "smtp" }, updateParts),
{ smtp: { host: "smtp.example.org" } },
"server edits do not replay credential fields"
);
assertDeepEqual(
mailProfileTargetedUpdatePayload({ kind: "credentials", protocol: "imap" }, updateParts),
{ credentials: { imap: { username: "imap-user", password: "imap-secret" } } },
"credential edits send only the selected protocol"
);
assertEqual(
mailProfileCreateCredentialsPayload({ username: null }, { password: "" }),
undefined,
"credential-free creates omit the credential object"
);
assertDeepEqual(
mailProfileCreateCredentialsPayload(
{ username: "smtp-user", password: null },
{ username: null, password: "imap-secret" }
),
{ smtp: { username: "smtp-user" }, imap: { password: "imap-secret" } },
"create payloads retain only explicitly populated credential fields"
);