diff --git a/docs/ADDRESS_MODULE_ARCHITECTURE.md b/docs/ADDRESS_MODULE_ARCHITECTURE.md index 36104d0..24820a5 100644 --- a/docs/ADDRESS_MODULE_ARCHITECTURE.md +++ b/docs/ADDRESS_MODULE_ARCHITECTURE.md @@ -176,6 +176,37 @@ module retirement audits all remaining owned credential material before table removal. An unowned legacy reference is detached rather than passed to an external secret provider. +LDAP and Active Directory use the same source, plan, diagnostic, tombstone, and +provider-health records. Endpoints must use LDAPS or StartTLS and may reference +only a visible reusable credential envelope; bind secrets are never copied into +source metadata. Root-DSE discovery returns candidate base DNs. A configured +source performs bounded paged searches and maps explicit attributes to contact +fields. Stable source keys plus `modifyTimestamp`, `uSNChanged`, `entryCSN`, or +a deterministic attribute digest make refreshes idempotent. Only a complete +successful search can infer deletion. A timeout, bind failure, malformed entry, +duplicate key, or configured entry limit retains existing contacts and reports +the source as failed/stale instead of creating tombstones. + +## Static Tabular Imports + +CSV and XLSX use versioned, scoped mapping profiles rather than live sync +sources. Profiles retain delimiter, encoding, header or worksheet selection, +stable source-key mapping, field mappings, locale and tags, row limits, and +explicit duplicate, blank-value, and existing-contact policies. Updating a +profile creates an immutable next version; prior import runs continue to point +at the reviewed version. + +Preview decodes at most 10 MB and 10,000 rows, validates every referenced +column and source key, and returns an effect or diagnostic for every data row. +XLSX parsing is read-only; formulas are rejected and macros/legacy workbook +formats are not accepted. The input SHA-256 and deterministic plan hash are +stored with full effects. Apply uses exactly that plan, rejects changed target +contacts, and is idempotent. Created IDs and pre-update snapshots provide a +guarded rollback: rollback proceeds only while each imported contact still +matches its recorded post-apply hash. Arbitrary transforms remain Dataflow's +responsibility; Files and Datasources are optional origins, not prerequisites +for direct upload. + ## Quality, Deduplication, And Recovery Quality is evidence about a concrete contact point, separate from communication @@ -208,15 +239,15 @@ Implement connectors in this order: 1. vCard import/export and batch import. 2. CardDAV address books. -3. [LDAP/Active Directory read-only directories](https://git.add-ideas.de/GovOPlaN/govoplan-addresses/issues/15) - and [reusable CSV/XLSX mapping profiles](https://git.add-ideas.de/GovOPlaN/govoplan-addresses/issues/19). +3. LDAP/Active Directory read-only directories and reusable CSV/XLSX mapping + profiles (implemented). 4. [Microsoft Graph for Microsoft 365](https://git.add-ideas.de/GovOPlaN/govoplan-addresses/issues/16), [explicit on-premises Exchange profiles](https://git.add-ideas.de/GovOPlaN/govoplan-addresses/issues/17), and [Google People](https://git.add-ideas.de/GovOPlaN/govoplan-addresses/issues/18). 5. [LDIF import](https://git.add-ideas.de/GovOPlaN/govoplan-addresses/issues/20) and [selective/large-batch vCard workflows](https://git.add-ideas.de/GovOPlaN/govoplan-addresses/issues/21). -The live connectors use the existing sync-source model. LDAP starts read-only; +The live connectors use the existing sync-source model. LDAP is read-only; Microsoft Graph and Google start with read-only/import and gate two-way mode on conditional-write and outcome-reconciliation tests. On-premises Exchange first probes and records an explicit supported server/API profile. CSV/XLSX, LDIF, diff --git a/pyproject.toml b/pyproject.toml index 94addc3..5ab3e3d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,8 @@ authors = [{ name = "GovOPlaN" }] dependencies = [ "defusedxml>=0.7.1", "govoplan-core>=0.1.11", + "ldap3>=2.9.1,<3", + "openpyxl>=3.1.5,<4", ] [tool.setuptools.packages.find] diff --git a/src/govoplan_addresses/backend/db/models.py b/src/govoplan_addresses/backend/db/models.py index 95ba733..c76f743 100644 --- a/src/govoplan_addresses/backend/db/models.py +++ b/src/govoplan_addresses/backend/db/models.py @@ -4,7 +4,7 @@ import uuid from datetime import datetime from typing import Any -from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, JSON, String, Text, text +from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, JSON, String, Text, UniqueConstraint, text from sqlalchemy.orm import Mapped, mapped_column, relationship from govoplan_core.db.base import Base, TimestampMixin @@ -549,8 +549,70 @@ class AddressSyncDiagnostic(Base, TimestampMixin): sync_source: Mapped[AddressSyncSource] = relationship(back_populates="diagnostics") +class AddressImportProfile(Base, TimestampMixin): + __tablename__ = "addresses_import_profiles" + __table_args__ = ( + UniqueConstraint("profile_key", "version", name="uq_addresses_import_profile_version"), + Index("ix_addresses_import_profiles_scope", "tenant_id", "scope_type", "scope_id", "is_current"), + Index("ix_addresses_import_profiles_format", "tenant_id", "source_format"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + profile_key: Mapped[str] = mapped_column(String(36), nullable=False, default=new_uuid, index=True) + version: Mapped[int] = mapped_column(Integer, nullable=False, default=1) + tenant_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + scope_type: Mapped[str] = mapped_column(String(20), nullable=False, index=True) + scope_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + name: Mapped[str] = mapped_column(String(255), nullable=False) + description: Mapped[str | None] = mapped_column(Text, nullable=True) + source_format: Mapped[str] = mapped_column(String(20), nullable=False, index=True) + configuration: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict) + is_current: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, index=True) + created_by_account_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + superseded_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + + +class AddressImportRun(Base, TimestampMixin): + __tablename__ = "addresses_import_runs" + __table_args__ = ( + Index("ix_addresses_import_runs_book_status", "address_book_id", "status", "created_at"), + Index("ix_addresses_import_runs_tenant_hash", "tenant_id", "input_hash"), + ) + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) + tenant_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + address_book_id: Mapped[str] = mapped_column( + ForeignKey("addresses_address_books.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + profile_id: Mapped[str] = mapped_column( + ForeignKey("addresses_import_profiles.id", ondelete="RESTRICT"), + nullable=False, + index=True, + ) + source_filename: Mapped[str] = mapped_column(String(500), nullable=False) + source_format: Mapped[str] = mapped_column(String(20), nullable=False, index=True) + input_hash: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + plan_hash: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + status: Mapped[str] = mapped_column(String(30), nullable=False, default="previewed", index=True) + row_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + statistics: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict) + diagnostics: Mapped[list[dict[str, Any]]] = mapped_column(JSON, nullable=False, default=list) + plan_data: Mapped[list[dict[str, Any]]] = mapped_column(JSON, nullable=False, default=list) + result_evidence: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict) + created_by_account_id: Mapped[str | None] = mapped_column(String(36), nullable=True, index=True) + applied_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + rolled_back_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + + address_book: Mapped[AddressBook] = relationship() + profile: Mapped[AddressImportProfile] = relationship() + + __all__ = [ "AddressBook", + "AddressImportProfile", + "AddressImportRun", "AddressList", "AddressListEntry", "AddressSyncConflict", diff --git a/src/govoplan_addresses/backend/import_schemas.py b/src/govoplan_addresses/backend/import_schemas.py new file mode 100644 index 0000000..e3c83a6 --- /dev/null +++ b/src/govoplan_addresses/backend/import_schemas.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +AddressImportFormat = Literal["csv", "xlsx"] +AddressImportScope = Literal["user", "group", "tenant", "system"] + +IMPORT_TARGET_FIELDS = frozenset( + { + "source_key", + "display_name", + "given_name", + "family_name", + "organization", + "role_title", + "note", + "email", + "phone", + "street", + "postal_code", + "locality", + "region", + "country", + "tags", + "visibility", + } +) + + +class AddressImportConfiguration(BaseModel): + field_mappings: dict[str, str] = Field(default_factory=dict, max_length=40) + delimiter: Literal[",", ";", "\t", "|"] = "," + encoding: Literal["utf-8", "utf-8-sig", "cp1252", "latin-1"] = "utf-8-sig" + header_row: int = Field(default=1, ge=1, le=100) + sheet_name: str | None = Field(default=None, max_length=255) + source_key_column: str | None = Field(default=None, max_length=255) + duplicate_source_key_policy: Literal["reject", "first", "last"] = "reject" + existing_contact_policy: Literal["update", "ignore", "reject"] = "update" + blank_value_policy: Literal["ignore", "clear", "reject"] = "ignore" + locale: str | None = Field(default=None, max_length=35) + default_tags: list[str] = Field(default_factory=list, max_length=100) + max_rows: int = Field(default=10_000, ge=1, le=10_000) + + @model_validator(mode="after") + def validate_mappings(self) -> "AddressImportConfiguration": + invalid = sorted(set(self.field_mappings).difference(IMPORT_TARGET_FIELDS)) + if invalid: + raise ValueError(f"Unsupported address import target fields: {', '.join(invalid)}") + for target, column in self.field_mappings.items(): + if not target.strip() or not column.strip(): + raise ValueError("Import mapping targets and source columns cannot be blank.") + if "source_key" not in self.field_mappings and not self.source_key_column: + raise ValueError("Address import profiles require a stable source-key column.") + return self + + +class AddressImportProfileCreateRequest(BaseModel): + scope_type: AddressImportScope = "tenant" + scope_id: str | None = Field(default=None, max_length=36) + name: str = Field(min_length=1, max_length=255) + description: str | None = Field(default=None, max_length=4000) + source_format: AddressImportFormat + configuration: AddressImportConfiguration + + +class AddressImportProfileUpdateRequest(BaseModel): + name: str | None = Field(default=None, min_length=1, max_length=255) + description: str | None = Field(default=None, max_length=4000) + configuration: AddressImportConfiguration | None = None + + +class AddressImportProfileResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: str + profile_key: str + version: int + tenant_id: str | None = None + scope_type: str + scope_id: str | None = None + name: str + description: str | None = None + source_format: str + configuration: AddressImportConfiguration + is_current: bool + created_by_account_id: str | None = None + superseded_at: datetime | None = None + created_at: datetime + updated_at: datetime + + +class AddressImportProfileListResponse(BaseModel): + profiles: list[AddressImportProfileResponse] = Field(default_factory=list) + + +class AddressImportFilePayload(BaseModel): + filename: str = Field(min_length=1, max_length=500) + content_base64: str = Field(min_length=1, max_length=14_000_000) + + +class AddressImportPreviewRequest(AddressImportFilePayload): + profile_id: str = Field(min_length=1, max_length=36) + + +class AddressImportEffectResponse(BaseModel): + row_number: int + action: Literal["create", "update", "conflict", "unchanged", "ignored"] + source_key: str | None = None + contact_id: str | None = None + display_name: str | None = None + changed_fields: list[str] = Field(default_factory=list) + message: str | None = None + + +class AddressImportDiagnosticResponse(BaseModel): + severity: Literal["info", "warning", "error"] + code: str + message: str + row_number: int | None = None + field: str | None = None + details: dict[str, Any] = Field(default_factory=dict) + + +class AddressImportRunResponse(BaseModel): + id: str + address_book_id: str + profile_id: str + source_filename: str + source_format: str + input_hash: str + plan_hash: str + status: str + row_count: int + statistics: dict[str, int] = Field(default_factory=dict) + diagnostics: list[AddressImportDiagnosticResponse] = Field(default_factory=list) + effects: list[AddressImportEffectResponse] = Field(default_factory=list) + can_apply: bool + result_evidence: dict[str, Any] = Field(default_factory=dict) + created_at: datetime + updated_at: datetime + applied_at: datetime | None = None + rolled_back_at: datetime | None = None + + +class AddressImportCommitRequest(BaseModel): + expected_plan_hash: str = Field(min_length=64, max_length=64) + + +class AddressImportRollbackRequest(BaseModel): + reason: str = Field(min_length=3, max_length=2000) + + +__all__ = [ + "AddressImportCommitRequest", + "AddressImportConfiguration", + "AddressImportDiagnosticResponse", + "AddressImportEffectResponse", + "AddressImportFilePayload", + "AddressImportFormat", + "AddressImportPreviewRequest", + "AddressImportProfileCreateRequest", + "AddressImportProfileListResponse", + "AddressImportProfileResponse", + "AddressImportProfileUpdateRequest", + "AddressImportRollbackRequest", + "AddressImportRunResponse", + "IMPORT_TARGET_FIELDS", +] diff --git a/src/govoplan_addresses/backend/imports.py b/src/govoplan_addresses/backend/imports.py new file mode 100644 index 0000000..5346302 --- /dev/null +++ b/src/govoplan_addresses/backend/imports.py @@ -0,0 +1,911 @@ +from __future__ import annotations + +import base64 +import binascii +import csv +import hashlib +import json +from collections import Counter +from io import BytesIO, StringIO +from typing import Any + +from sqlalchemy import and_, false, or_ +from sqlalchemy.orm import Session + +from govoplan_addresses.backend.db.models import ( + AddressImportProfile, + AddressImportRun, + Contact, +) +from govoplan_addresses.backend.import_schemas import ( + AddressImportConfiguration, + AddressImportPreviewRequest, + AddressImportProfileCreateRequest, + AddressImportProfileUpdateRequest, + AddressImportRollbackRequest, +) +from govoplan_addresses.backend.schemas import ( + ContactCreateRequest, + ContactEmailPayload, + ContactPhonePayload, + ContactPostalAddressPayload, + ContactUpdateRequest, +) +from govoplan_addresses.backend.service import ( + AddressBookError, + create_contact, + delete_contact, + get_visible_address_book, + get_visible_contact, + restore_contact, + update_contact, +) +from govoplan_core.auth import ApiPrincipal +from govoplan_core.db.base import utcnow + + +MAX_IMPORT_BYTES = 10_000_000 +MAX_IMPORT_COLUMNS = 200 + + +def _account_id(principal: ApiPrincipal) -> str: + return principal.account_id + + +def _tenant_id(principal: ApiPrincipal) -> str: + return principal.tenant_id + + +def _profile_scope_predicate(principal: ApiPrincipal): + tenant_id = _tenant_id(principal) + predicates = [AddressImportProfile.scope_type == "system"] + predicates.extend( + [ + and_(AddressImportProfile.tenant_id == tenant_id, AddressImportProfile.scope_type == "tenant"), + and_( + AddressImportProfile.tenant_id == tenant_id, + AddressImportProfile.scope_type == "user", + AddressImportProfile.scope_id == _account_id(principal), + ), + ] + ) + group_ids = tuple(principal.group_ids) + if group_ids: + predicates.append( + and_( + AddressImportProfile.tenant_id == tenant_id, + AddressImportProfile.scope_type == "group", + AddressImportProfile.scope_id.in_(group_ids), + ) + ) + return or_(*predicates) if predicates else false() + + +def list_import_profiles( + session: Session, + principal: ApiPrincipal, + *, + include_history: bool = False, +) -> list[AddressImportProfile]: + query = session.query(AddressImportProfile).filter(_profile_scope_predicate(principal)) + if not include_history: + query = query.filter(AddressImportProfile.is_current.is_(True)) + return query.order_by(AddressImportProfile.name.asc(), AddressImportProfile.version.desc()).all() + + +def get_import_profile( + session: Session, + principal: ApiPrincipal, + profile_id: str, +) -> AddressImportProfile: + profile = ( + session.query(AddressImportProfile) + .filter(_profile_scope_predicate(principal), AddressImportProfile.id == profile_id) + .one_or_none() + ) + if profile is None: + raise AddressBookError("Address import profile not found.") + return profile + + +def create_import_profile( + session: Session, + principal: ApiPrincipal, + payload: AddressImportProfileCreateRequest, +) -> AddressImportProfile: + tenant_id, scope_id = _validated_profile_scope(principal, payload.scope_type, payload.scope_id) + profile = AddressImportProfile( + tenant_id=tenant_id, + scope_type=payload.scope_type, + scope_id=scope_id, + name=payload.name.strip(), + description=_trim(payload.description), + source_format=payload.source_format, + configuration=payload.configuration.model_dump(mode="json"), + is_current=True, + created_by_account_id=_account_id(principal), + ) + session.add(profile) + return profile + + +def update_import_profile( + session: Session, + principal: ApiPrincipal, + profile_id: str, + payload: AddressImportProfileUpdateRequest, +) -> AddressImportProfile: + current = get_import_profile(session, principal, profile_id) + if not current.is_current: + raise AddressBookError("Only the current import profile version can be updated.") + current.is_current = False + current.superseded_at = utcnow() + next_profile = AddressImportProfile( + profile_key=current.profile_key, + version=current.version + 1, + tenant_id=current.tenant_id, + scope_type=current.scope_type, + scope_id=current.scope_id, + name=(payload.name.strip() if payload.name is not None else current.name), + description=(payload.description.strip() or None if payload.description is not None else current.description), + source_format=current.source_format, + configuration=( + payload.configuration.model_dump(mode="json") + if payload.configuration is not None + else dict(current.configuration or {}) + ), + is_current=True, + created_by_account_id=_account_id(principal), + ) + session.add(next_profile) + return next_profile + + +def retire_import_profile( + session: Session, + principal: ApiPrincipal, + profile_id: str, +) -> None: + profile = get_import_profile(session, principal, profile_id) + if profile.scope_type == "system" and not principal.has("addresses:address_book:admin"): + raise AddressBookError("System import profiles require address-book administration permission.") + profile.is_current = False + profile.superseded_at = utcnow() + + +def preview_address_import( + session: Session, + principal: ApiPrincipal, + address_book_id: str, + payload: AddressImportPreviewRequest, +) -> AddressImportRun: + book = get_visible_address_book(session, principal, address_book_id) + if book.read_only: + raise AddressBookError("Static imports require a writable address book.") + profile = get_import_profile(session, principal, payload.profile_id) + raw = _decode_payload(payload.content_base64) + input_hash = hashlib.sha256(raw).hexdigest() + config = AddressImportConfiguration.model_validate(profile.configuration) + rows, parse_diagnostics = _parse_rows( + raw, + filename=payload.filename, + source_format=profile.source_format, + config=config, + ) + plan_data, map_diagnostics = _plan_rows( + session, + book_id=book.id, + profile=profile, + input_hash=input_hash, + rows=rows, + config=config, + ) + diagnostics = [*parse_diagnostics, *map_diagnostics] + statistics = dict(Counter(item["action"] for item in plan_data)) + statistics["rows"] = len(rows) + statistics["errors"] = sum(item["severity"] == "error" for item in diagnostics) + statistics["warnings"] = sum(item["severity"] == "warning" for item in diagnostics) + plan_hash = _hash_json( + { + "profile_id": profile.id, + "profile_version": profile.version, + "address_book_id": book.id, + "input_hash": input_hash, + "plan": plan_data, + } + ) + run = AddressImportRun( + tenant_id=book.tenant_id, + address_book_id=book.id, + profile_id=profile.id, + source_filename=payload.filename.strip(), + source_format=profile.source_format, + input_hash=input_hash, + plan_hash=plan_hash, + status="previewed", + row_count=len(rows), + statistics=statistics, + diagnostics=diagnostics, + plan_data=plan_data, + result_evidence={}, + created_by_account_id=_account_id(principal), + ) + session.add(run) + session.flush() + return run + + +def get_import_run( + session: Session, + principal: ApiPrincipal, + run_id: str, +) -> AddressImportRun: + visible_book_ids = [book.id for book in _visible_import_books(session, principal)] + if not visible_book_ids: + raise AddressBookError("Address import run not found.") + run = ( + session.query(AddressImportRun) + .filter(AddressImportRun.id == run_id, AddressImportRun.address_book_id.in_(visible_book_ids)) + .one_or_none() + ) + if run is None: + raise AddressBookError("Address import run not found.") + return run + + +def apply_address_import( + session: Session, + principal: ApiPrincipal, + run_id: str, + *, + expected_plan_hash: str, +) -> AddressImportRun: + run = get_import_run(session, principal, run_id) + if run.status == "applied": + return run + if run.status != "previewed": + raise AddressBookError(f"Import run cannot be applied from status {run.status!r}.") + if run.plan_hash != expected_plan_hash: + raise AddressBookError("The reviewed import plan changed; create a new preview.") + if any(item.get("severity") == "error" for item in run.diagnostics or []): + raise AddressBookError("Import plans with error diagnostics cannot be applied.") + if any(item.get("action") == "conflict" for item in run.plan_data or []): + raise AddressBookError("Resolve import conflicts by correcting the file or mapping profile and preview again.") + + created_ids: list[str] = [] + updated: list[dict[str, Any]] = [] + for item in run.plan_data or []: + action = item.get("action") + if action in {"ignored", "unchanged"}: + continue + source_ref = str(item["source_ref"]) + existing = _contact_by_source_ref(session, run.address_book_id, source_ref) + if action == "create": + if existing is not None and existing.deleted_at is None: + raise AddressBookError("A target contact appeared after preview; preview the import again.") + contact = create_contact( + session, + principal, + run.address_book_id, + ContactCreateRequest.model_validate(item["payload"]), + ) + _stamp_import_contact(contact, run=run, item=item) + session.flush() + created_ids.append(contact.id) + item["contact_id"] = contact.id + item["after_hash"] = _contact_hash(contact) + elif action == "update": + if existing is None: + raise AddressBookError("An import target disappeared after preview; preview the import again.") + if _contact_hash(existing) != item.get("expected_contact_hash"): + raise AddressBookError( + f'Contact "{existing.display_name}" changed after preview; preview the import again.' + ) + before = _contact_snapshot(existing) + if existing.deleted_at is not None: + restore_contact(session, principal, existing.id) + contact = update_contact( + session, + principal, + existing.id, + ContactUpdateRequest.model_validate(item["payload"]), + ) + _stamp_import_contact(contact, run=run, item=item) + session.flush() + updated.append({"contact_id": contact.id, "before": before, "after_hash": _contact_hash(contact)}) + item["contact_id"] = contact.id + + run.status = "applied" + run.applied_at = utcnow() + run.plan_data = list(run.plan_data or []) + run.result_evidence = { + "input_hash": run.input_hash, + "plan_hash": run.plan_hash, + "created_contact_ids": created_ids, + "updated_contacts": updated, + "applied_by_account_id": _account_id(principal), + "applied_at": run.applied_at.isoformat(), + } + return run + + +def rollback_address_import( + session: Session, + principal: ApiPrincipal, + run_id: str, + payload: AddressImportRollbackRequest, +) -> AddressImportRun: + run = get_import_run(session, principal, run_id) + if run.status == "rolled_back": + return run + if run.status != "applied": + raise AddressBookError("Only an applied import can be rolled back.") + evidence = dict(run.result_evidence or {}) + updated = list(evidence.get("updated_contacts") or []) + created_ids = list(evidence.get("created_contact_ids") or []) + + expected_hashes = { + str(item["contact_id"]): str(item["after_hash"]) + for item in updated + } + expected_hashes.update( + { + str(item["contact_id"]): str(item["after_hash"]) + for item in run.plan_data or [] + if item.get("contact_id") in created_ids and item.get("after_hash") + } + ) + for contact_id, expected_hash in expected_hashes.items(): + contact = get_visible_contact(session, principal, contact_id, include_deleted=True) + if _contact_hash(contact) != expected_hash: + raise AddressBookError( + f'Contact "{contact.display_name}" changed after import; automatic rollback is unsafe.' + ) + + for contact_id in created_ids: + contact = get_visible_contact(session, principal, contact_id, include_deleted=True) + if contact.deleted_at is None: + delete_contact(session, principal, contact.id) + for item in updated: + contact = get_visible_contact(session, principal, str(item["contact_id"]), include_deleted=True) + snapshot = dict(item["before"]) + if contact.deleted_at is not None: + restore_contact(session, principal, contact.id) + update_contact( + session, + principal, + contact.id, + ContactUpdateRequest.model_validate(snapshot["payload"]), + ) + contact.source_kind = snapshot.get("source_kind") or "local" + contact.source_ref = snapshot.get("source_ref") + contact.source_revision = snapshot.get("source_revision") + contact.source_payload_kind = snapshot.get("source_payload_kind") + contact.source_payload_raw = snapshot.get("source_payload_raw") + contact.provenance = dict(snapshot.get("provenance") or {}) + contact.metadata_ = dict(snapshot.get("metadata") or {}) + + run.status = "rolled_back" + run.rolled_back_at = utcnow() + run.result_evidence = { + **evidence, + "rollback_reason": payload.reason, + "rolled_back_by_account_id": _account_id(principal), + "rolled_back_at": run.rolled_back_at.isoformat(), + } + return run + + +def import_run_payload(run: AddressImportRun) -> dict[str, Any]: + diagnostics = list(run.diagnostics or []) + effects = [ + { + "row_number": int(item["row_number"]), + "action": item["action"], + "source_key": item.get("source_key"), + "contact_id": item.get("contact_id"), + "display_name": item.get("display_name"), + "changed_fields": list(item.get("changed_fields") or []), + "message": item.get("message"), + } + for item in run.plan_data or [] + ] + can_apply = ( + run.status == "previewed" + and not any(item.get("severity") == "error" for item in diagnostics) + and not any(item.get("action") == "conflict" for item in run.plan_data or []) + ) + evidence = dict(run.result_evidence or {}) + public_evidence = { + key: evidence[key] + for key in ( + "input_hash", + "plan_hash", + "applied_by_account_id", + "applied_at", + "rollback_reason", + "rolled_back_by_account_id", + "rolled_back_at", + ) + if evidence.get(key) is not None + } + if evidence: + public_evidence["created_contact_count"] = len(evidence.get("created_contact_ids") or []) + public_evidence["updated_contact_count"] = len(evidence.get("updated_contacts") or []) + + return { + "id": run.id, + "address_book_id": run.address_book_id, + "profile_id": run.profile_id, + "source_filename": run.source_filename, + "source_format": run.source_format, + "input_hash": run.input_hash, + "plan_hash": run.plan_hash, + "status": run.status, + "row_count": run.row_count, + "statistics": dict(run.statistics or {}), + "diagnostics": diagnostics, + "effects": effects, + "can_apply": can_apply, + # Full before-images remain private rollback evidence and must not be + # projected through a normal import-run read response. + "result_evidence": public_evidence, + "created_at": run.created_at, + "updated_at": run.updated_at, + "applied_at": run.applied_at, + "rolled_back_at": run.rolled_back_at, + } + + +def _validated_profile_scope( + principal: ApiPrincipal, + scope_type: str, + requested_scope_id: str | None, +) -> tuple[str | None, str | None]: + if scope_type == "system": + if not principal.has("addresses:address_book:admin"): + raise AddressBookError("System import profiles require address-book administration permission.") + return None, None + tenant_id = _tenant_id(principal) + if scope_type == "tenant": + return tenant_id, tenant_id + if scope_type == "user": + return tenant_id, _account_id(principal) + if scope_type == "group": + scope_id = _trim(requested_scope_id) + if scope_id is None: + raise AddressBookError("Group import profiles require a group id.") + if scope_id not in principal.group_ids and not principal.has("addresses:address_book:admin"): + raise AddressBookError("The selected group is not visible to the current principal.") + return tenant_id, scope_id + raise AddressBookError("Unsupported import profile scope.") + + +def _decode_payload(encoded: str) -> bytes: + try: + raw = base64.b64decode(encoded, validate=True) + except (binascii.Error, ValueError) as exc: + raise AddressBookError("Import file content is not valid base64.") from exc + if not raw: + raise AddressBookError("Import file is empty.") + if len(raw) > MAX_IMPORT_BYTES: + raise AddressBookError(f"Import files are limited to {MAX_IMPORT_BYTES} bytes.") + return raw + + +def _parse_rows( + raw: bytes, + *, + filename: str, + source_format: str, + config: AddressImportConfiguration, +) -> tuple[list[tuple[int, dict[str, str]]], list[dict[str, Any]]]: + if source_format == "csv": + return _parse_csv(raw, config=config) + if source_format == "xlsx": + if not filename.casefold().endswith(".xlsx"): + raise AddressBookError("XLSX imports require an .xlsx file; macros and legacy workbooks are not accepted.") + return _parse_xlsx(raw, config=config) + raise AddressBookError(f"Unsupported address import format: {source_format!r}.") + + +def _parse_csv( + raw: bytes, + *, + config: AddressImportConfiguration, +) -> tuple[list[tuple[int, dict[str, str]]], list[dict[str, Any]]]: + try: + text = raw.decode(config.encoding) + except UnicodeDecodeError as exc: + raise AddressBookError(f"CSV is not valid {config.encoding}: {exc}.") from exc + reader = csv.reader(StringIO(text), delimiter=config.delimiter) + all_rows = list(reader) + if len(all_rows) < config.header_row: + raise AddressBookError("CSV does not contain the configured header row.") + header = _headers(all_rows[config.header_row - 1]) + result: list[tuple[int, dict[str, str]]] = [] + for row_number, values in enumerate(all_rows[config.header_row :], start=config.header_row + 1): + if not any(str(value).strip() for value in values): + continue + if len(values) > MAX_IMPORT_COLUMNS: + raise AddressBookError(f"CSV row {row_number} exceeds the {MAX_IMPORT_COLUMNS}-column limit.") + result.append((row_number, _row_dict(header, values))) + if len(result) > config.max_rows: + raise AddressBookError(f"CSV exceeds the configured {config.max_rows}-row limit.") + return result, [] + + +def _parse_xlsx( + raw: bytes, + *, + config: AddressImportConfiguration, +) -> tuple[list[tuple[int, dict[str, str]]], list[dict[str, Any]]]: + try: + from openpyxl import load_workbook + except ImportError as exc: # pragma: no cover - dependency/package failure + raise AddressBookError("XLSX import support is not installed.") from exc + try: + workbook = load_workbook(BytesIO(raw), read_only=True, data_only=False, keep_links=False) + except Exception as exc: + raise AddressBookError(f"XLSX workbook could not be read: {exc}.") from exc + if len(workbook.sheetnames) > 100: + raise AddressBookError("XLSX workbooks are limited to 100 sheets.") + if config.sheet_name: + if config.sheet_name not in workbook.sheetnames: + raise AddressBookError(f'XLSX sheet "{config.sheet_name}" was not found.') + sheet = workbook[config.sheet_name] + else: + sheet = workbook[workbook.sheetnames[0]] + rows = list(sheet.iter_rows(min_row=config.header_row, max_row=config.header_row)) + if not rows: + raise AddressBookError("XLSX does not contain the configured header row.") + header = _headers([cell.value for cell in rows[0]]) + result: list[tuple[int, dict[str, str]]] = [] + for row_number, cells in enumerate(sheet.iter_rows(min_row=config.header_row + 1), start=config.header_row + 1): + if len(cells) > MAX_IMPORT_COLUMNS: + raise AddressBookError(f"XLSX row {row_number} exceeds the {MAX_IMPORT_COLUMNS}-column limit.") + if any(cell.data_type == "f" for cell in cells): + raise AddressBookError(f"XLSX row {row_number} contains a formula; formulas are never evaluated during import.") + values = [cell.value for cell in cells] + if not any(value is not None and str(value).strip() for value in values): + continue + result.append((row_number, _row_dict(header, values))) + if len(result) > config.max_rows: + raise AddressBookError(f"XLSX exceeds the configured {config.max_rows}-row limit.") + return result, [] + + +def _headers(values: list[Any]) -> list[str]: + headers = [str(value).strip() if value is not None else "" for value in values] + if not headers or not any(headers): + raise AddressBookError("Import header row is empty.") + if len(headers) > MAX_IMPORT_COLUMNS: + raise AddressBookError(f"Import files are limited to {MAX_IMPORT_COLUMNS} columns.") + blank = [index + 1 for index, value in enumerate(headers) if not value] + if blank: + raise AddressBookError(f"Import header contains blank column names at positions {blank}.") + duplicates = sorted(name for name, count in Counter(headers).items() if count > 1) + if duplicates: + raise AddressBookError(f"Import header contains duplicate columns: {', '.join(duplicates)}.") + return headers + + +def _row_dict(headers: list[str], values: list[Any]) -> dict[str, str]: + padded = [*values, *([None] * max(0, len(headers) - len(values)))] + return { + header: "" if value is None else str(value).strip() + for header, value in zip(headers, padded, strict=False) + } + + +def _plan_rows( + session: Session, + *, + book_id: str, + profile: AddressImportProfile, + input_hash: str, + rows: list[tuple[int, dict[str, str]]], + config: AddressImportConfiguration, +) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + diagnostics: list[dict[str, Any]] = [] + plan: list[dict[str, Any]] = [] + headers = set(rows[0][1]) if rows else set() + referenced_columns = set(config.field_mappings.values()) + if config.source_key_column: + referenced_columns.add(config.source_key_column) + missing_columns = sorted(referenced_columns.difference(headers)) + for column in missing_columns: + diagnostics.append(_diagnostic("error", "missing_column", f'Configured column "{column}" is missing.', field=column)) + if missing_columns: + return [], diagnostics + + key_column = config.source_key_column or config.field_mappings["source_key"] + keyed_rows: list[tuple[int, dict[str, str], str]] = [] + key_counts: Counter[str] = Counter() + for row_number, row in rows: + key = row.get(key_column, "").strip() + if not key: + diagnostics.append(_diagnostic("error", "missing_source_key", "Stable source key is blank.", row_number=row_number, field=key_column)) + plan.append(_plan_effect(row_number, "conflict", source_key=None, message="Stable source key is blank.")) + continue + key_counts[key] += 1 + keyed_rows.append((row_number, row, key)) + + first_index: dict[str, int] = {} + last_index: dict[str, int] = {} + for index, (_row_number, _row, key) in enumerate(keyed_rows): + first_index.setdefault(key, index) + last_index[key] = index + + for index, (row_number, row, key) in enumerate(keyed_rows): + if key_counts[key] > 1: + if config.duplicate_source_key_policy == "reject": + diagnostics.append(_diagnostic("error", "duplicate_source_key", f'Duplicate source key "{key}".', row_number=row_number, field=key_column)) + plan.append(_plan_effect(row_number, "conflict", source_key=key, message="Duplicate source key.")) + continue + chosen = first_index[key] if config.duplicate_source_key_policy == "first" else last_index[key] + if index != chosen: + diagnostics.append(_diagnostic("warning", "duplicate_source_key_ignored", f'Duplicate source key "{key}" was ignored by profile policy.', row_number=row_number, field=key_column)) + plan.append(_plan_effect(row_number, "ignored", source_key=key, message="Duplicate row ignored by profile policy.")) + continue + + mapped, row_diagnostics = _mapped_fields(row_number, row, config=config) + diagnostics.extend(row_diagnostics) + source_ref = f"import:{profile.profile_key}:{key}" + existing = _contact_by_source_ref(session, book_id, source_ref) + payload = _payload_from_mapped(mapped, profile=profile, input_hash=input_hash, row_number=row_number, source_key=key) + display_name = payload.get("display_name") or payload.get("email") or key + if any(item["severity"] == "error" for item in row_diagnostics): + plan.append(_plan_effect(row_number, "conflict", source_key=key, display_name=display_name, source_ref=source_ref, payload=payload, message="Row validation failed.")) + continue + if existing is None: + plan.append(_plan_effect(row_number, "create", source_key=key, display_name=display_name, source_ref=source_ref, payload=payload, changed_fields=sorted(mapped))) + continue + if config.existing_contact_policy == "ignore": + plan.append(_plan_effect(row_number, "ignored", source_key=key, contact_id=existing.id, display_name=existing.display_name, source_ref=source_ref, payload=payload, message="Existing contact retained by profile policy.")) + continue + if config.existing_contact_policy == "reject": + diagnostics.append(_diagnostic("error", "existing_contact", f'Contact for source key "{key}" already exists.', row_number=row_number)) + plan.append(_plan_effect(row_number, "conflict", source_key=key, contact_id=existing.id, display_name=existing.display_name, source_ref=source_ref, payload=payload, message="Existing contact rejected by profile policy.")) + continue + changed_fields = _changed_fields(existing, mapped) + plan.append( + _plan_effect( + row_number, + "update" if changed_fields or existing.deleted_at is not None else "unchanged", + source_key=key, + contact_id=existing.id, + display_name=display_name, + source_ref=source_ref, + payload=payload, + changed_fields=changed_fields, + expected_contact_hash=_contact_hash(existing), + ) + ) + return sorted(plan, key=lambda item: item["row_number"]), diagnostics + + +def _mapped_fields( + row_number: int, + row: dict[str, str], + *, + config: AddressImportConfiguration, +) -> tuple[dict[str, Any], list[dict[str, Any]]]: + mapped: dict[str, Any] = {} + diagnostics: list[dict[str, Any]] = [] + for target, column in config.field_mappings.items(): + if target == "source_key": + continue + value = row.get(column, "").strip() + if not value: + if config.blank_value_policy == "reject": + diagnostics.append(_diagnostic("error", "blank_value", f'Column "{column}" is blank.', row_number=row_number, field=target)) + elif config.blank_value_policy == "clear": + mapped[target] = [] if target == "tags" else None + continue + mapped[target] = [item.strip() for item in value.split(",") if item.strip()] if target == "tags" else value + if config.default_tags: + mapped["tags"] = list(dict.fromkeys([*(mapped.get("tags") or []), *config.default_tags])) + if not any(mapped.get(name) for name in ("display_name", "given_name", "family_name", "email", "organization")): + diagnostics.append(_diagnostic("error", "missing_identity", "Row has no name, email, or organization to identify the contact.", row_number=row_number)) + return mapped, diagnostics + + +def _payload_from_mapped( + mapped: dict[str, Any], + *, + profile: AddressImportProfile, + input_hash: str, + row_number: int, + source_key: str, +) -> dict[str, Any]: + display_name = mapped.get("display_name") or " ".join(filter(None, [mapped.get("given_name"), mapped.get("family_name")])) or mapped.get("email") or mapped.get("organization") + payload: dict[str, Any] = { + key: mapped.get(key) + for key in ("given_name", "family_name", "organization", "role_title", "note", "tags") + if key in mapped + } + payload["display_name"] = display_name + if "email" in mapped: + payload["emails"] = [] if mapped["email"] is None else [ContactEmailPayload(email=mapped["email"], is_primary=True).model_dump(mode="json")] + if "phone" in mapped: + payload["phones"] = [] if mapped["phone"] is None else [ContactPhonePayload(phone=mapped["phone"], is_primary=True).model_dump(mode="json")] + postal_keys = {"street", "postal_code", "locality", "region", "country"} + if postal_keys.intersection(mapped): + postal = {key: mapped.get(key) for key in postal_keys if key in mapped} + payload["postal_addresses"] = [ContactPostalAddressPayload(**postal, is_primary=True).model_dump(mode="json")] if any(postal.values()) else [] + payload["provenance"] = { + "import": { + "profile_key": profile.profile_key, + "profile_id": profile.id, + "profile_version": profile.version, + "input_hash": input_hash, + "row_number": row_number, + "source_key": source_key, + "locale": profile.configuration.get("locale"), + "visibility": mapped.get("visibility"), + } + } + return ContactCreateRequest.model_validate(payload).model_dump( + mode="json", + exclude_unset=True, + exclude_none=False, + ) + + +def _changed_fields(contact: Contact, mapped: dict[str, Any]) -> list[str]: + current: dict[str, Any] = { + "display_name": contact.display_name, + "given_name": contact.given_name, + "family_name": contact.family_name, + "organization": contact.organization, + "role_title": contact.role_title, + "note": contact.note, + "tags": list(contact.tags or []), + "email": contact.emails[0].email if contact.emails else None, + "phone": contact.phones[0].phone if contact.phones else None, + } + if contact.postal_addresses: + postal = contact.postal_addresses[0] + current.update({key: getattr(postal, key) for key in ("street", "postal_code", "locality", "region", "country")}) + return sorted(key for key, value in mapped.items() if key != "visibility" and current.get(key) != value) + + +def _plan_effect( + row_number: int, + action: str, + *, + source_key: str | None, + contact_id: str | None = None, + display_name: str | None = None, + source_ref: str | None = None, + payload: dict[str, Any] | None = None, + changed_fields: list[str] | None = None, + message: str | None = None, + expected_contact_hash: str | None = None, +) -> dict[str, Any]: + return { + "row_number": row_number, + "action": action, + "source_key": source_key, + "contact_id": contact_id, + "display_name": display_name, + "source_ref": source_ref, + "payload": payload or {}, + "changed_fields": changed_fields or [], + "message": message, + "expected_contact_hash": expected_contact_hash, + } + + +def _diagnostic( + severity: str, + code: str, + message: str, + *, + row_number: int | None = None, + field: str | None = None, +) -> dict[str, Any]: + return { + "severity": severity, + "code": code, + "message": message, + "row_number": row_number, + "field": field, + "details": {}, + } + + +def _contact_by_source_ref(session: Session, book_id: str, source_ref: str) -> Contact | None: + return ( + session.query(Contact) + .filter(Contact.address_book_id == book_id, Contact.source_ref == source_ref) + .order_by(Contact.created_at.asc(), Contact.id.asc()) + .first() + ) + + +def _stamp_import_contact(contact: Contact, *, run: AddressImportRun, item: dict[str, Any]) -> None: + contact.source_kind = run.source_format + contact.source_ref = item["source_ref"] + contact.source_revision = hashlib.sha256( + f'{run.input_hash}:{item["row_number"]}:{item["source_key"]}'.encode() + ).hexdigest() + contact.source_payload_kind = f"{run.source_format}-mapped-row" + contact.source_payload_raw = None + provenance = dict(contact.provenance or {}) + provenance["import_run_id"] = run.id + provenance["input_hash"] = run.input_hash + provenance["plan_hash"] = run.plan_hash + contact.provenance = provenance + + +def _contact_snapshot(contact: Contact) -> dict[str, Any]: + return { + "payload": { + "display_name": contact.display_name, + "given_name": contact.given_name, + "family_name": contact.family_name, + "organization": contact.organization, + "role_title": contact.role_title, + "note": contact.note, + "tags": list(contact.tags or []), + "emails": [{"label": item.label, "email": item.email, "is_primary": item.is_primary} for item in contact.emails], + "phones": [{"label": item.label, "phone": item.phone, "is_primary": item.is_primary} for item in contact.phones], + "postal_addresses": [ + { + "label": item.label, + "street": item.street, + "postal_code": item.postal_code, + "locality": item.locality, + "region": item.region, + "country": item.country, + "is_primary": item.is_primary, + } + for item in contact.postal_addresses + ], + "provenance": dict(contact.provenance or {}), + }, + "source_kind": contact.source_kind, + "source_ref": contact.source_ref, + "source_revision": contact.source_revision, + "source_payload_kind": contact.source_payload_kind, + "source_payload_raw": contact.source_payload_raw, + "provenance": dict(contact.provenance or {}), + "metadata": dict(contact.metadata_ or {}), + } + + +def _contact_hash(contact: Contact) -> str: + return _hash_json({**_contact_snapshot(contact), "deleted_at": contact.deleted_at.isoformat() if contact.deleted_at else None}) + + +def _hash_json(value: object) -> str: + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()).hexdigest() + + +def _trim(value: str | None) -> str | None: + if value is None: + return None + normalized = value.strip() + return normalized or None + + +def _visible_import_books(session: Session, principal: ApiPrincipal): + from govoplan_addresses.backend.service import list_address_books + + return list_address_books(session, principal) + + +__all__ = [ + "apply_address_import", + "create_import_profile", + "get_import_profile", + "get_import_run", + "import_run_payload", + "list_import_profiles", + "preview_address_import", + "retire_import_profile", + "rollback_address_import", + "update_import_profile", +] diff --git a/src/govoplan_addresses/backend/ldap.py b/src/govoplan_addresses/backend/ldap.py new file mode 100644 index 0000000..575c9e2 --- /dev/null +++ b/src/govoplan_addresses/backend/ldap.py @@ -0,0 +1,196 @@ +from __future__ import annotations + +import ssl +from dataclasses import dataclass +from typing import Any +from urllib.parse import unquote, urlsplit + +from govoplan_core.core.connector_runtime import ConnectorContractError, ConnectorEndpoint + + +class AddressLdapError(RuntimeError): + pass + + +@dataclass(frozen=True, slots=True) +class AddressLdapEntry: + dn: str + attributes: dict[str, Any] + + +@dataclass(frozen=True, slots=True) +class AddressLdapSearchResult: + base_dn: str + entries: tuple[AddressLdapEntry, ...] + complete: bool + page_size: int + + +class AddressLdapClient: + def __init__( + self, + *, + url: str, + bind_dn: str | None = None, + password: str | None = None, + start_tls: bool = True, + connect_timeout: int = 10, + receive_timeout: int = 30, + ) -> None: + try: + endpoint = ConnectorEndpoint( + url=url, + tls_mode="start_tls" if start_tls else "required", + ) + except ConnectorContractError as exc: + raise AddressLdapError(str(exc)) from exc + parsed = urlsplit(endpoint.url) + if parsed.scheme not in {"ldap", "ldaps"}: + raise AddressLdapError("LDAP endpoints must use ldap:// or ldaps://.") + if parsed.scheme == "ldap" and not start_tls: + raise AddressLdapError("ldap:// endpoints require StartTLS.") + if parsed.path not in {"", "/"}: + self.default_base_dn = unquote(parsed.path.lstrip("/")) + else: + self.default_base_dn = None + self.url = endpoint.url + self.host = parsed.hostname or "" + self.port = parsed.port or (636 if parsed.scheme == "ldaps" else 389) + self.use_ssl = parsed.scheme == "ldaps" + self.start_tls = parsed.scheme == "ldap" and start_tls + self.bind_dn = bind_dn + self.password = password + self.connect_timeout = max(1, min(connect_timeout, 30)) + self.receive_timeout = max(1, min(receive_timeout, 120)) + + def discover_base_dns(self) -> tuple[str, ...]: + connection = self._connection() + try: + from ldap3 import BASE + + if not connection.search( + search_base="", + search_filter="(objectClass=*)", + search_scope=BASE, + attributes=["namingContexts", "defaultNamingContext", "rootDomainNamingContext"], + ): + raise AddressLdapError(_ldap_result_message(connection.result, "LDAP root DSE discovery failed.")) + values: list[str] = [] + for entry in connection.entries: + data = entry.entry_attributes_as_dict + for key in ("defaultNamingContext", "rootDomainNamingContext", "namingContexts"): + for value in _as_values(data.get(key)): + normalized = str(value).strip() + if normalized and normalized not in values: + values.append(normalized) + if self.default_base_dn and self.default_base_dn not in values: + values.insert(0, self.default_base_dn) + return tuple(values) + finally: + connection.unbind() + + def search( + self, + *, + base_dn: str, + search_filter: str, + attributes: tuple[str, ...], + page_size: int = 500, + max_entries: int = 10_000, + ) -> AddressLdapSearchResult: + normalized_base = base_dn.strip() or self.default_base_dn + if not normalized_base: + raise AddressLdapError("LDAP base DN is required.") + page_size = max(1, min(page_size, 1_000)) + max_entries = max(1, min(max_entries, 10_000)) + connection = self._connection() + entries: list[AddressLdapEntry] = [] + complete = True + try: + try: + stream = connection.extend.standard.paged_search( + search_base=normalized_base, + search_filter=search_filter, + attributes=list(attributes), + paged_size=page_size, + generator=True, + ) + for response in stream: + response_type = response.get("type") + if response_type != "searchResEntry": + continue + if len(entries) >= max_entries: + complete = False + break + entries.append( + AddressLdapEntry( + dn=str(response.get("dn") or ""), + attributes=dict(response.get("attributes") or {}), + ) + ) + except Exception as exc: + raise AddressLdapError(f"LDAP paged search failed: {exc}.") from exc + if connection.result and int(connection.result.get("result", 0) or 0) != 0: + raise AddressLdapError(_ldap_result_message(connection.result, "LDAP paged search failed.")) + return AddressLdapSearchResult( + base_dn=normalized_base, + entries=tuple(entries), + complete=complete, + page_size=page_size, + ) + finally: + connection.unbind() + + def _connection(self): + try: + from ldap3 import Connection, Server, Tls + except ImportError as exc: # pragma: no cover - package failure + raise AddressLdapError("LDAP connector support is not installed.") from exc + tls = Tls(validate=ssl.CERT_REQUIRED, version=ssl.PROTOCOL_TLS_CLIENT) + server = Server( + self.host, + port=self.port, + use_ssl=self.use_ssl, + tls=tls, + connect_timeout=self.connect_timeout, + ) + try: + connection = Connection( + server, + user=self.bind_dn, + password=self.password, + receive_timeout=self.receive_timeout, + raise_exceptions=True, + ) + connection.open() + if self.start_tls: + connection.start_tls() + connection.bind() + return connection + except Exception as exc: + raise AddressLdapError(f"LDAP connection or bind failed: {exc}.") from exc + + +def _as_values(value: Any) -> tuple[Any, ...]: + if value is None: + return () + if isinstance(value, (list, tuple, set)): + return tuple(value) + return (value,) + + +def _ldap_result_message(result: dict[str, Any] | None, fallback: str) -> str: + if not result: + return fallback + description = str(result.get("description") or "").strip() + message = str(result.get("message") or "").strip() + detail = ": ".join(part for part in (description, message) if part) + return f"{fallback} {detail}".strip() + + +__all__ = [ + "AddressLdapClient", + "AddressLdapEntry", + "AddressLdapError", + "AddressLdapSearchResult", +] diff --git a/src/govoplan_addresses/backend/ldap_schemas.py b/src/govoplan_addresses/backend/ldap_schemas.py new file mode 100644 index 0000000..6a38c6e --- /dev/null +++ b/src/govoplan_addresses/backend/ldap_schemas.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, Field, model_validator + + +DEFAULT_LDAP_ATTRIBUTE_MAP: dict[str, str] = { + "source_key": "entryUUID", + "source_revision": "modifyTimestamp", + "display_name": "displayName", + "given_name": "givenName", + "family_name": "sn", + "organization": "o", + "role_title": "title", + "email": "mail", + "phone": "telephoneNumber", + "street": "streetAddress", + "postal_code": "postalCode", + "locality": "l", + "region": "st", + "country": "c", + "tags": "memberOf", +} + + +class AddressLdapConnectionRequest(BaseModel): + url: str = Field(min_length=1, max_length=2000) + credential_ref: str | None = Field(default=None, max_length=1000) + bind_dn: str | None = Field(default=None, max_length=1000) + start_tls: bool = True + connect_timeout: int = Field(default=10, ge=1, le=30) + receive_timeout: int = Field(default=30, ge=1, le=120) + + +class AddressLdapDiscoveryResponse(BaseModel): + base_dns: list[str] = Field(default_factory=list) + + +class AddressLdapSourceCreateRequest(AddressLdapConnectionRequest): + display_name: str = Field(min_length=1, max_length=255) + base_dn: str = Field(min_length=1, max_length=2000) + search_filter: str = Field(default="(&(objectClass=person)(mail=*))", min_length=1, max_length=2000) + page_size: int = Field(default=500, ge=1, le=1000) + max_entries: int = Field(default=10_000, ge=1, le=10_000) + attribute_map: dict[str, str] = Field(default_factory=lambda: dict(DEFAULT_LDAP_ATTRIBUTE_MAP), max_length=40) + + @model_validator(mode="after") + def validate_mapping(self) -> "AddressLdapSourceCreateRequest": + if "source_key" not in self.attribute_map: + raise ValueError("LDAP mappings require a stable source_key attribute.") + if not any(key in self.attribute_map for key in ("display_name", "email", "given_name", "family_name", "organization")): + raise ValueError("LDAP mappings require at least one contact identity attribute.") + if any(not key.strip() or not value.strip() for key, value in self.attribute_map.items()): + raise ValueError("LDAP mapping names and attributes cannot be blank.") + return self + + +class AddressLdapTestResponse(BaseModel): + success: bool + base_dn: str + sampled_entries: int + attributes: list[str] = Field(default_factory=list) + diagnostic: dict[str, Any] = Field(default_factory=dict) + + +__all__ = [ + "AddressLdapConnectionRequest", + "AddressLdapDiscoveryResponse", + "AddressLdapSourceCreateRequest", + "AddressLdapTestResponse", + "DEFAULT_LDAP_ATTRIBUTE_MAP", +] diff --git a/src/govoplan_addresses/backend/manifest.py b/src/govoplan_addresses/backend/manifest.py index 3ecdd8d..913a2ad 100644 --- a/src/govoplan_addresses/backend/manifest.py +++ b/src/govoplan_addresses/backend/manifest.py @@ -38,11 +38,15 @@ from govoplan_core.core.provider_governance import ( from govoplan_core.db.base import Base from govoplan_addresses.backend.provider_state import ( CARDDAV_PROVIDER_ID, + LDAP_PROVIDER_ID, carddav_provider_states, + ldap_provider_states, ) _addresses_table_retirement_provider = drop_table_retirement_provider( + addresses_models.AddressImportRun, + addresses_models.AddressImportProfile, addresses_models.ContactFieldProvenance, addresses_models.ContactRedirect, addresses_models.ContactMergeRecord, @@ -155,6 +159,8 @@ ROLE_TEMPLATES = ( def _tenant_summary(session, tenant_id: str) -> dict[str, int]: from govoplan_addresses.backend.db.models import ( AddressBook, + AddressImportProfile, + AddressImportRun, AddressList, AddressSyncSource, Contact, @@ -171,6 +177,8 @@ def _tenant_summary(session, tenant_id: str) -> dict[str, int]: "contact_quality_decisions": session.query(ContactPointQualityDecision).filter(ContactPointQualityDecision.tenant_id == tenant_id).count(), "contact_point_snapshots": session.query(ContactPointSnapshot).filter(ContactPointSnapshot.tenant_id == tenant_id).count(), "sync_sources": session.query(AddressSyncSource).filter(AddressSyncSource.tenant_id == tenant_id, AddressSyncSource.enabled.is_(True)).count(), + "address_import_profiles": session.query(AddressImportProfile).filter(AddressImportProfile.tenant_id == tenant_id, AddressImportProfile.is_current.is_(True)).count(), + "address_import_runs": session.query(AddressImportRun).filter(AddressImportRun.tenant_id == tenant_id).count(), } @@ -233,12 +241,59 @@ CARDDAV_PROVIDER = ExternalProviderDeclaration( ) +LDAP_PROVIDER = ExternalProviderDeclaration( + id=LDAP_PROVIDER_ID, + module_id="addresses", + label="Read-only LDAP and Active Directory contacts", + maturity="synchronize", + operations=("discover", "read", "preview", "synchronize"), + objects=( + ProviderObjectDeclaration( + object_type="contact", + field_groups=("identity", "name", "organization", "postal", "email", "phone", "source_metadata"), + authority_modes=("external_authoritative", "external_mirror"), + default_authority_mode="external_authoritative", + ), + ), + behavior=ProviderBehaviorDeclaration( + revision_tokens="Stable LDAP source keys plus modifyTimestamp, uSNChanged, entryCSN, or a deterministic attribute digest are retained.", + concurrency="LDAP is authoritative and read-only; local projections are replaced only from a complete reviewed plan.", + freshness="Last attempt, last success, remote revision, and stale provider health remain visible.", + health="TLS, bind, discovery, paging, mapping, truncation, and malformed-entry failures are separate diagnostics.", + max_read_items=10000, + idempotency="The source binding, stable key, and revision prevent duplicate contact projections.", + retry="Failed reads are retried only by a new operator or scheduled sync attempt with bounded timeouts.", + timeout_seconds=120, + conflicts="Duplicate source keys, malformed mappings, and locally changed projections block or require a fresh plan.", + outcome_unknown="Read failures never infer external deletions and retain prior local projections as stale.", + outcome_unknown_supported=True, + evidence="Source keys, revisions, mapping configuration, diagnostics, tombstones, and normalized field provenance are retained.", + audit_event_types=( + "addresses.sync_source_created", + "addresses.sync_previewed", + "addresses.sync_completed", + ), + correction="Correct the directory or mapping, then run a new full preview and synchronization.", + rollback="Prior projections remain reconstructable from source revision and contact change evidence; external LDAP is never mutated.", + compensation="A later authoritative refresh restores corrected projections.", + reconciliation="Only a complete paged search may infer an absent source object and create a local tombstone.", + outage="Existing contacts remain available and visibly stale; an unavailable directory never causes deletes.", + classifications=("personal", "confidential", "restricted"), + purposes=("directory projection", "recipient resolution", "identity-linked contact discovery"), + retention="Address, audit, and records policies govern local projections and tombstone evidence.", + secret_handling="Bind secrets remain in reusable credential envelopes; URLs, previews, and diagnostics contain no credentials.", + ), + capability_names=(CAPABILITY_ADDRESSES_LOOKUP, CAPABILITY_ADDRESSES_CONTACT_WRITER), + documentation_topic_ids=("addresses.ldap-directory",), +) + + manifest = ModuleManifest( id="addresses", name="Addresses", version="0.1.9", required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), - optional_dependencies=("campaigns", "mail", "forms", "reporting", "portal", "postbox"), + optional_dependencies=("campaigns", "mail", "forms", "reporting", "portal", "postbox", "connectors"), provides_interfaces=( ModuleInterfaceProvider(name=CAPABILITY_ADDRESSES_LOOKUP, version="0.1.8"), ModuleInterfaceProvider(name=CAPABILITY_ADDRESSES_PEOPLE_SEARCH, version="0.1.0"), @@ -288,6 +343,8 @@ manifest = ModuleManifest( }, uninstall_guard_providers=( persistent_table_uninstall_guard( + addresses_models.AddressImportRun, + addresses_models.AddressImportProfile, addresses_models.AddressSyncDiagnostic, addresses_models.AddressSyncConflict, addresses_models.AddressSyncTombstone, @@ -340,6 +397,40 @@ manifest = ModuleManifest( related_modules=("dist_lists", "campaigns", "policy", "templates"), order=31, ), + DocumentationTopic( + id="addresses.tabular-imports", + title="CSV and XLSX contact imports", + summary="Preview and apply reusable, versioned contact mappings without silent row loss.", + body=( + "CSV and XLSX files can be mapped with scoped, reusable profile versions. Each preview validates headers, " + "encodings, source keys, duplicates, blank values, workbook limits, and contact identity before any mutation. " + "The reviewed input hash and plan hash are retained with row-level effects and diagnostics. Apply is idempotent, " + "rejects contacts changed after preview, and records sufficient evidence for a guarded rollback. XLSX formulas, " + "macros, and legacy workbook formats are never executed or imported." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("tenant_admin", "module_admin", "power_user"), + related_modules=("connectors", "datasources", "dataflow", "files", "audit"), + order=33, + ), + DocumentationTopic( + id="addresses.ldap-directory", + title="LDAP and Active Directory address sources", + summary="Project authoritative directory contacts through a bounded, read-only synchronization source.", + body=( + "LDAP sources use LDAPS or StartTLS and reusable credential envelopes. Discovery finds available base DNs; " + "the source profile then controls a bounded paged filter and explicit attribute mapping. Preview never mutates " + "contacts. A complete successful read may create, update, or tombstone local projections; truncated or failed " + "reads suppress absence-based deletes and mark the source stale. Stable source keys, revisions, normalized fields, " + "and provenance remain attached to every retained contact." + ), + layer="configured", + documentation_types=("admin", "user"), + audience=("tenant_admin", "operator", "module_admin"), + related_modules=("connectors", "idm", "access", "policy", "audit"), + order=34, + ), DocumentationTopic( id="addresses.quality-and-merge", title="Contact quality, duplicates, and reversible merges", @@ -360,13 +451,18 @@ manifest = ModuleManifest( order=32, ), ), - external_providers=(CARDDAV_PROVIDER,), + external_providers=(CARDDAV_PROVIDER, LDAP_PROVIDER), external_provider_state_providers=( ExternalProviderStateProviderRegistration( module_id="addresses", provider_id=CARDDAV_PROVIDER_ID, provider=carddav_provider_states, ), + ExternalProviderStateProviderRegistration( + module_id="addresses", + provider_id=LDAP_PROVIDER_ID, + provider=ldap_provider_states, + ), ), architecture=declared_module_architecture( layer="communication_participation", diff --git a/src/govoplan_addresses/backend/migrations/versions/c5d7e8f9a0b1_address_import_profiles.py b/src/govoplan_addresses/backend/migrations/versions/c5d7e8f9a0b1_address_import_profiles.py new file mode 100644 index 0000000..7c7dfa6 --- /dev/null +++ b/src/govoplan_addresses/backend/migrations/versions/c5d7e8f9a0b1_address_import_profiles.py @@ -0,0 +1,95 @@ +"""add versioned address import profiles and immutable run evidence + +Revision ID: c5d7e8f9a0b1 +Revises: b4c6d7e8f9a0 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "c5d7e8f9a0b1" +down_revision = "b4c6d7e8f9a0" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "addresses_import_profiles", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("profile_key", sa.String(length=36), nullable=False), + sa.Column("version", sa.Integer(), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=True), + sa.Column("scope_type", sa.String(length=20), nullable=False), + sa.Column("scope_id", sa.String(length=36), nullable=True), + sa.Column("name", sa.String(length=255), nullable=False), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("source_format", sa.String(length=20), nullable=False), + sa.Column("configuration", sa.JSON(), nullable=False), + sa.Column("is_current", sa.Boolean(), nullable=False), + sa.Column("created_by_account_id", sa.String(length=36), nullable=True), + sa.Column("superseded_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("profile_key", "version", name="uq_addresses_import_profile_version"), + ) + op.create_index("ix_addresses_import_profiles_profile_key", "addresses_import_profiles", ["profile_key"]) + op.create_index("ix_addresses_import_profiles_tenant_id", "addresses_import_profiles", ["tenant_id"]) + op.create_index("ix_addresses_import_profiles_scope_type", "addresses_import_profiles", ["scope_type"]) + op.create_index("ix_addresses_import_profiles_scope_id", "addresses_import_profiles", ["scope_id"]) + op.create_index("ix_addresses_import_profiles_source_format", "addresses_import_profiles", ["source_format"]) + op.create_index("ix_addresses_import_profiles_is_current", "addresses_import_profiles", ["is_current"]) + op.create_index("ix_addresses_import_profiles_created_by_account_id", "addresses_import_profiles", ["created_by_account_id"]) + op.create_index("ix_addresses_import_profiles_superseded_at", "addresses_import_profiles", ["superseded_at"]) + op.create_index("ix_addresses_import_profiles_scope", "addresses_import_profiles", ["tenant_id", "scope_type", "scope_id", "is_current"]) + op.create_index("ix_addresses_import_profiles_format", "addresses_import_profiles", ["tenant_id", "source_format"]) + + op.create_table( + "addresses_import_runs", + sa.Column("id", sa.String(length=36), nullable=False), + sa.Column("tenant_id", sa.String(length=36), nullable=True), + sa.Column("address_book_id", sa.String(length=36), nullable=False), + sa.Column("profile_id", sa.String(length=36), nullable=False), + sa.Column("source_filename", sa.String(length=500), nullable=False), + sa.Column("source_format", sa.String(length=20), nullable=False), + sa.Column("input_hash", sa.String(length=64), nullable=False), + sa.Column("plan_hash", sa.String(length=64), nullable=False), + sa.Column("status", sa.String(length=30), nullable=False), + sa.Column("row_count", sa.Integer(), nullable=False), + sa.Column("statistics", sa.JSON(), nullable=False), + sa.Column("diagnostics", sa.JSON(), nullable=False), + sa.Column("plan_data", sa.JSON(), nullable=False), + sa.Column("result_evidence", sa.JSON(), nullable=False), + sa.Column("created_by_account_id", sa.String(length=36), nullable=True), + sa.Column("applied_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("rolled_back_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["address_book_id"], ["addresses_address_books.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["profile_id"], ["addresses_import_profiles.id"], ondelete="RESTRICT"), + sa.PrimaryKeyConstraint("id"), + ) + for column in ( + "tenant_id", + "address_book_id", + "profile_id", + "source_format", + "input_hash", + "plan_hash", + "status", + "created_by_account_id", + "applied_at", + "rolled_back_at", + ): + op.create_index(f"ix_addresses_import_runs_{column}", "addresses_import_runs", [column]) + op.create_index("ix_addresses_import_runs_book_status", "addresses_import_runs", ["address_book_id", "status", "created_at"]) + op.create_index("ix_addresses_import_runs_tenant_hash", "addresses_import_runs", ["tenant_id", "input_hash"]) + + +def downgrade() -> None: + op.drop_table("addresses_import_runs") + op.drop_table("addresses_import_profiles") diff --git a/src/govoplan_addresses/backend/provider_state.py b/src/govoplan_addresses/backend/provider_state.py index 2bf142f..3663d16 100644 --- a/src/govoplan_addresses/backend/provider_state.py +++ b/src/govoplan_addresses/backend/provider_state.py @@ -18,16 +18,45 @@ from govoplan_core.core.provider_governance import ( CARDDAV_PROVIDER_ID = "addresses.carddav_sync" +LDAP_PROVIDER_ID = "addresses.ldap_directory" _CURRENT_WINDOW = timedelta(hours=24) def carddav_provider_states( context: ExternalProviderStateContext, +) -> tuple[ExternalProviderRuntimeState, ...]: + if not isinstance(context.session, Session): + raise RuntimeError("Addresses provider state requires a database session.") + return _provider_states( + context, + connector_types=("carddav",), + provider_id=CARDDAV_PROVIDER_ID, + label="CardDAV", + ) + + +def ldap_provider_states( + context: ExternalProviderStateContext, +) -> tuple[ExternalProviderRuntimeState, ...]: + return _provider_states( + context, + connector_types=("ldap", "active_directory"), + provider_id=LDAP_PROVIDER_ID, + label="LDAP/Active Directory", + ) + + +def _provider_states( + context: ExternalProviderStateContext, + *, + connector_types: tuple[str, ...], + provider_id: str, + label: str, ) -> tuple[ExternalProviderRuntimeState, ...]: if not isinstance(context.session, Session): raise RuntimeError("Addresses provider state requires a database session.") statement = select(AddressSyncSource).where( - AddressSyncSource.connector_type == "carddav" + AddressSyncSource.connector_type.in_(connector_types) ) if context.tenant_id is not None: statement = statement.where(AddressSyncSource.tenant_id == context.tenant_id) @@ -58,6 +87,8 @@ def carddav_provider_states( return tuple( _source_state( source, + provider_id=provider_id, + label=label, observed_at=observed_at, conflict_count=conflict_counts.get(source.id, 0), error_count=error_counts.get(source.id, 0), @@ -86,6 +117,8 @@ def _grouped_counts( def _source_state( source: AddressSyncSource, *, + provider_id: str, + label: str, observed_at: datetime, conflict_count: int, error_count: int, @@ -113,10 +146,14 @@ def _source_state( else "ready" ) return ExternalProviderRuntimeState( - provider_id=CARDDAV_PROVIDER_ID, + provider_id=provider_id, binding_ref=f"addresses:sync-source:{source.id}", authority_mode=( - "external_mirror" if source.read_only else "governed_sync" + "external_authoritative" + if provider_id == LDAP_PROVIDER_ID + else "external_mirror" + if source.read_only + else "governed_sync" ), observed_at=observed_at, configured=True, @@ -127,13 +164,13 @@ def _source_state( recovery=recovery, last_success_at=_aware(source.last_success_at), detail=( - "CardDAV source is disabled." + f"{label} source is disabled." if not active - else "CardDAV source requires reconciliation." + else f"{label} source requires reconciliation." if conflict == "pending" - else "CardDAV source health has not been observed yet." + else f"{label} source health has not been observed yet." if health == "unknown" - else "CardDAV source state is available." + else f"{label} source state is available." ), metrics={ "open_conflicts": conflict_count, @@ -161,4 +198,9 @@ def _aware(value: datetime | None) -> datetime | None: return value.astimezone(UTC) -__all__ = ["CARDDAV_PROVIDER_ID", "carddav_provider_states"] +__all__ = [ + "CARDDAV_PROVIDER_ID", + "LDAP_PROVIDER_ID", + "carddav_provider_states", + "ldap_provider_states", +] diff --git a/src/govoplan_addresses/backend/router.py b/src/govoplan_addresses/backend/router.py index 07670fb..c54f33b 100644 --- a/src/govoplan_addresses/backend/router.py +++ b/src/govoplan_addresses/backend/router.py @@ -15,6 +15,12 @@ from govoplan_core.core.contact_points import ( from govoplan_core.core.distribution_lists import DistributionSourceReference from govoplan_core.db.session import get_session from govoplan_addresses.backend.carddav import AddressCardDAVError +from govoplan_addresses.backend.ldap import AddressLdapError +from govoplan_addresses.backend.ldap_schemas import ( + AddressLdapConnectionRequest, + AddressLdapDiscoveryResponse, + AddressLdapSourceCreateRequest, +) from govoplan_addresses.backend.db.models import ( AddressBook, AddressList, @@ -29,6 +35,27 @@ from govoplan_addresses.backend.db.models import ( ContactPointQualityDecision, ContactPostalAddress, ) +from govoplan_addresses.backend.import_schemas import ( + AddressImportCommitRequest, + AddressImportPreviewRequest, + AddressImportProfileCreateRequest, + AddressImportProfileListResponse, + AddressImportProfileResponse, + AddressImportProfileUpdateRequest, + AddressImportRollbackRequest, + AddressImportRunResponse, +) +from govoplan_addresses.backend.imports import ( + apply_address_import, + create_import_profile, + get_import_run, + import_run_payload, + list_import_profiles, + preview_address_import, + retire_import_profile, + rollback_address_import, + update_import_profile, +) from govoplan_addresses.backend.capabilities import ( AddressesContactPointResolutionCapability, AddressesContactWriterCapability, @@ -112,6 +139,7 @@ from govoplan_addresses.backend.service import ( create_address_list, create_address_list_entry, create_carddav_sync_source, + create_ldap_sync_source, create_contact, create_contact_channel_rule, create_contact_quality_decision, @@ -124,6 +152,7 @@ from govoplan_addresses.backend.service import ( delete_contact, delete_sync_source, discover_carddav_address_books, + discover_ldap_base_dns, export_address_book_vcard, export_contact_vcard, end_contact_channel_rule, @@ -1341,6 +1370,55 @@ def api_discover_carddav_address_books( raise _error(AddressBookError(str(exc))) from exc +@router.post("/ldap/discover", response_model=AddressLdapDiscoveryResponse) +def api_discover_ldap_base_dns( + payload: AddressLdapConnectionRequest, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:sync:write") + try: + return AddressLdapDiscoveryResponse( + base_dns=list(discover_ldap_base_dns(session, principal, payload)) + ) + except (AddressBookError, AddressLdapError) as exc: + raise _error(AddressBookError(str(exc))) from exc + + +@router.post( + "/address-books/{book_id}/ldap/sources", + response_model=AddressSyncSourceResponse, + status_code=status.HTTP_201_CREATED, +) +def api_create_ldap_sync_source( + book_id: str, + payload: AddressLdapSourceCreateRequest, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:sync:write") + try: + sync_source = create_ldap_sync_source(session, principal, book_id, payload) + _audit_address_sync( + session, + principal, + action="addresses.sync_source_created", + object_type="address_sync_source", + object_id=sync_source.id, + details={ + "address_book_id": book_id, + "connector_type": "ldap", + "sync_direction": "read_only", + }, + ) + session.commit() + session.refresh(sync_source) + return _sync_source_response(sync_source) + except (AddressBookError, AddressLdapError) as exc: + session.rollback() + raise _error(AddressBookError(str(exc))) from exc + + @router.get("/credentials", response_model=AddressCredentialEnvelopeListResponse) def api_list_address_credentials( source_id: str | None = Query(default=None), @@ -1381,7 +1459,7 @@ def api_create_carddav_sync_source( session.commit() session.refresh(sync_source) return _sync_source_response(sync_source) - except (AddressBookError, AddressCardDAVError) as exc: + except (AddressBookError, AddressCardDAVError, AddressLdapError) as exc: session.rollback() raise _error(AddressBookError(str(exc))) from exc @@ -1428,7 +1506,7 @@ def api_preview_sync_source( ) session.commit() return _sync_plan_response(plan) - except (AddressBookError, AddressCardDAVError) as exc: + except (AddressBookError, AddressCardDAVError, AddressLdapError) as exc: session.rollback() raise _error(AddressBookError(str(exc))) from exc @@ -1460,9 +1538,41 @@ def api_run_sync_source( ) session.commit() return _sync_plan_response(plan) - except (AddressBookError, AddressCardDAVError) as exc: + except (AddressBookError, AddressCardDAVError, AddressLdapError) as exc: + message = str(exc) session.rollback() - raise _error(AddressBookError(str(exc))) from exc + try: + finish_sync_attempt( + session, + principal, + sync_source_id, + AddressSyncAttemptFinishRequest( + status="failed", + error=message, + diagnostic={ + "severity": "error", + "code": "connector_unavailable", + "message": message, + "retryable": True, + "stage": "read", + }, + ), + ) + record_sync_diagnostic( + session, + principal, + sync_source_id, + AddressSyncDiagnosticCreateRequest( + severity="error", + code="connector_unavailable", + message=message, + details={"retryable": True, "stage": "read"}, + ), + ) + session.commit() + except Exception: + session.rollback() + raise _error(AddressBookError(message)) from exc @router.post("/address-books/{book_id}/sync-sources", response_model=AddressSyncSourceResponse, status_code=status.HTTP_201_CREATED) @@ -1981,6 +2091,196 @@ def api_import_address_book_vcards( raise _error(AddressBookError(str(exc))) from exc +@router.get("/import-profiles", response_model=AddressImportProfileListResponse) +def api_list_address_import_profiles( + include_history: bool = Query(default=False), + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:contact:read") + return AddressImportProfileListResponse( + profiles=[ + AddressImportProfileResponse.model_validate(item) + for item in list_import_profiles(session, principal, include_history=include_history) + ] + ) + + +@router.post("/import-profiles", response_model=AddressImportProfileResponse, status_code=status.HTTP_201_CREATED) +def api_create_address_import_profile( + payload: AddressImportProfileCreateRequest, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:contact:write") + try: + profile = create_import_profile(session, principal, payload) + session.flush() + audit_from_principal( + session, + principal, + action="addresses.import_profile_created", + object_type="address_import_profile", + object_id=profile.profile_key, + details={"version": profile.version, "source_format": profile.source_format, "scope_type": profile.scope_type}, + ) + session.commit() + session.refresh(profile) + return AddressImportProfileResponse.model_validate(profile) + except AddressBookError as exc: + session.rollback() + raise _error(exc) from exc + + +@router.patch("/import-profiles/{profile_id}", response_model=AddressImportProfileResponse) +def api_update_address_import_profile( + profile_id: str, + payload: AddressImportProfileUpdateRequest, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:contact:write") + try: + profile = update_import_profile(session, principal, profile_id, payload) + session.flush() + audit_from_principal( + session, + principal, + action="addresses.import_profile_versioned", + object_type="address_import_profile", + object_id=profile.profile_key, + details={"version": profile.version, "source_format": profile.source_format}, + ) + session.commit() + session.refresh(profile) + return AddressImportProfileResponse.model_validate(profile) + except AddressBookError as exc: + session.rollback() + raise _error(exc) from exc + + +@router.delete("/import-profiles/{profile_id}", status_code=status.HTTP_204_NO_CONTENT) +def api_retire_address_import_profile( + profile_id: str, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:contact:write") + try: + retire_import_profile(session, principal, profile_id) + audit_from_principal( + session, + principal, + action="addresses.import_profile_retired", + object_type="address_import_profile", + object_id=profile_id, + ) + session.commit() + return Response(status_code=status.HTTP_204_NO_CONTENT) + except AddressBookError as exc: + session.rollback() + raise _error(exc) from exc + + +@router.post( + "/address-books/{book_id}/imports/preview", + response_model=AddressImportRunResponse, + status_code=status.HTTP_201_CREATED, +) +def api_preview_address_import( + book_id: str, + payload: AddressImportPreviewRequest, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:contact:write") + try: + run = preview_address_import(session, principal, book_id, payload) + audit_from_principal( + session, + principal, + action="addresses.import_previewed", + object_type="address_import_run", + object_id=run.id, + details={"input_hash": run.input_hash, "plan_hash": run.plan_hash, "statistics": run.statistics}, + ) + session.commit() + session.refresh(run) + return AddressImportRunResponse.model_validate(import_run_payload(run)) + except AddressBookError as exc: + session.rollback() + raise _error(exc) from exc + + +@router.get("/imports/{run_id}", response_model=AddressImportRunResponse) +def api_get_address_import_run( + run_id: str, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:contact:read") + try: + return AddressImportRunResponse.model_validate( + import_run_payload(get_import_run(session, principal, run_id)) + ) + except AddressBookError as exc: + raise _error(exc) from exc + + +@router.post("/imports/{run_id}/apply", response_model=AddressImportRunResponse) +def api_apply_address_import( + run_id: str, + payload: AddressImportCommitRequest, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:contact:write") + try: + run = apply_address_import(session, principal, run_id, expected_plan_hash=payload.expected_plan_hash) + session.flush() + audit_from_principal( + session, + principal, + action="addresses.import_applied", + object_type="address_import_run", + object_id=run.id, + details={"input_hash": run.input_hash, "plan_hash": run.plan_hash, "statistics": run.statistics}, + ) + session.commit() + session.refresh(run) + return AddressImportRunResponse.model_validate(import_run_payload(run)) + except AddressBookError as exc: + session.rollback() + raise _error(exc) from exc + + +@router.post("/imports/{run_id}/rollback", response_model=AddressImportRunResponse) +def api_rollback_address_import( + run_id: str, + payload: AddressImportRollbackRequest, + principal: ApiPrincipal = Depends(get_api_principal), + session: Session = Depends(get_session), +): + _require_scope(principal, "addresses:contact:write") + try: + run = rollback_address_import(session, principal, run_id, payload) + session.flush() + audit_from_principal( + session, + principal, + action="addresses.import_rolled_back", + object_type="address_import_run", + object_id=run.id, + details={"reason": payload.reason, "plan_hash": run.plan_hash}, + ) + session.commit() + session.refresh(run) + return AddressImportRunResponse.model_validate(import_run_payload(run)) + except AddressBookError as exc: + session.rollback() + raise _error(exc) from exc + + @router.get("/address-books/{book_id}/vcards/export") def api_export_address_book_vcards( book_id: str, diff --git a/src/govoplan_addresses/backend/service.py b/src/govoplan_addresses/backend/service.py index 5e5bfa3..a5aa010 100644 --- a/src/govoplan_addresses/backend/service.py +++ b/src/govoplan_addresses/backend/service.py @@ -38,6 +38,14 @@ from govoplan_addresses.backend.carddav import ( AddressCardDAVSyncUnsupported, ensure_collection_url, ) +from govoplan_addresses.backend.ldap import ( + AddressLdapClient, + AddressLdapError, +) +from govoplan_addresses.backend.ldap_schemas import ( + AddressLdapConnectionRequest, + AddressLdapSourceCreateRequest, +) from govoplan_addresses.backend.db.models import ( AddressBook, AddressList, @@ -142,6 +150,8 @@ class AddressSyncPlanItem: raw_vcard: str | None = None parsed_payload: ContactCreateRequest | None = None source_revision: str | None = None + raw_payload: str | None = None + source_details: dict[str, Any] = field(default_factory=dict) @dataclass(slots=True) @@ -386,6 +396,8 @@ def create_sync_source( raise AddressBookError("Sync source display name is required.") if connector_type.casefold() == "carddav" and not trusted_connector_metadata: _assert_api_carddav_metadata_safe(payload.metadata) + if connector_type.casefold() in {"ldap", "active_directory"} and not trusted_connector_metadata: + _assert_api_ldap_metadata_safe(payload.metadata) read_only = _read_only_from_sync_direction(payload.sync_direction, payload.read_only) sync_source = AddressSyncSource( tenant_id=book.tenant_id, @@ -449,6 +461,9 @@ def update_sync_source( if sync_source.connector_type.casefold() == "carddav": _assert_api_carddav_metadata_safe(metadata) metadata = _merge_server_owned_carddav_metadata(sync_source.metadata_, metadata) + elif sync_source.connector_type.casefold() in {"ldap", "active_directory"}: + _assert_api_ldap_metadata_safe(metadata) + metadata = _merge_server_owned_ldap_metadata(sync_source.metadata_, metadata) sync_source.metadata_ = metadata sync_source.updated_by_account_id = _account_id(principal) _apply_sync_source_to_book(sync_source.address_book, sync_source) @@ -774,7 +789,7 @@ def _apply_payload_sync_conflict(session: Session, principal: ApiPrincipal, conf parsed_payload=payload, source_revision=str(metadata.get("source_revision") or remote_value.get("etag") or "") or None, ) - _upsert_carddav_contact(session, principal, conflict.sync_source, item) + _upsert_remote_contact(session, principal, conflict.sync_source, item) def get_visible_sync_conflict(session: Session, principal: ApiPrincipal, conflict_id: str) -> AddressSyncConflict: @@ -849,6 +864,80 @@ def create_carddav_sync_source( return source +def discover_ldap_base_dns( + session: Session, + principal: ApiPrincipal, + payload: AddressLdapConnectionRequest, + *, + client: AddressLdapClient | None = None, +) -> tuple[str, ...]: + _assert_reusable_credential_ref(payload.credential_ref) + ldap_client = client or _ldap_client_from_connection_payload( + session, + principal, + payload, + ) + return ldap_client.discover_base_dns() + + +def create_ldap_sync_source( + session: Session, + principal: ApiPrincipal, + address_book_id: str, + payload: AddressLdapSourceCreateRequest, +) -> AddressSyncSource: + _assert_reusable_credential_ref(payload.credential_ref) + # Constructor validation rejects plaintext LDAP and embedded URL credentials + # without opening a network connection. + AddressLdapClient( + url=payload.url, + bind_dn=payload.bind_dn, + start_tls=payload.start_tls, + connect_timeout=payload.connect_timeout, + receive_timeout=payload.receive_timeout, + ) + metadata = { + "ldap": { + "url": payload.url.strip(), + "base_dn": payload.base_dn.strip(), + "search_filter": payload.search_filter.strip(), + "start_tls": payload.start_tls, + "connect_timeout": payload.connect_timeout, + "receive_timeout": payload.receive_timeout, + "page_size": payload.page_size, + "max_entries": payload.max_entries, + "attribute_map": dict(payload.attribute_map), + "bind_dn": _trim(payload.bind_dn), + "credential_ref": _trim(payload.credential_ref), + } + } + source = create_sync_source( + session, + principal, + address_book_id, + AddressSyncSourceCreateRequest( + connector_type="ldap", + display_name=payload.display_name, + external_account_ref=payload.url, + external_address_book_ref=payload.base_dn, + sync_direction="read_only", + read_only=True, + metadata=metadata, + ), + trusted_connector_metadata=True, + ) + reusable = _resolve_core_address_credential( + session, + tenant_id=principal.tenant_id, + source_id=source.id, + credential_ref=payload.credential_ref, + ) + if reusable is not None and not metadata["ldap"].get("bind_dn"): + metadata["ldap"]["bind_dn"] = _credential_username(reusable) + source.metadata_ = metadata + return source + + def preview_sync_source( session: Session, principal: ApiPrincipal, @@ -857,20 +946,27 @@ def preview_sync_source( force_full: bool = False, password: str | None = None, bearer_token: str | None = None, - client: AddressCardDAVClient | None = None, + client: AddressCardDAVClient | AddressLdapClient | None = None, ) -> AddressSyncPlan: sync_source = get_visible_sync_source(session, principal, sync_source_id) - if sync_source.connector_type != "carddav": - raise AddressBookError(f"Preview is not implemented for {sync_source.connector_type} sync sources.") - return _build_carddav_sync_plan( - session, - principal, - sync_source, - force_full=force_full, - password=password, - bearer_token=bearer_token, - client=client, - ) + if sync_source.connector_type == "carddav": + return _build_carddav_sync_plan( + session, + principal, + sync_source, + force_full=force_full, + password=password, + bearer_token=bearer_token, + client=client, # type: ignore[arg-type] + ) + if sync_source.connector_type in {"ldap", "active_directory"}: + return _build_ldap_sync_plan( + session, + principal, + sync_source, + client=client, # type: ignore[arg-type] + ) + raise AddressBookError(f"Preview is not implemented for {sync_source.connector_type} sync sources.") def run_sync_source( @@ -881,23 +977,38 @@ def run_sync_source( force_full: bool = False, password: str | None = None, bearer_token: str | None = None, - client: AddressCardDAVClient | None = None, + client: AddressCardDAVClient | AddressLdapClient | None = None, ) -> AddressSyncPlan: sync_source = start_sync_attempt(session, principal, sync_source_id) write_client = client if sync_source.connector_type == "carddav" and write_client is None: write_client = _carddav_client_for_source(session, sync_source, password=password, bearer_token=bearer_token) try: - plan = _build_carddav_sync_plan( + if sync_source.connector_type == "carddav": + plan = _build_carddav_sync_plan( + session, + principal, + sync_source, + force_full=force_full, + password=password, + bearer_token=bearer_token, + client=write_client, # type: ignore[arg-type] + ) + elif sync_source.connector_type in {"ldap", "active_directory"}: + plan = _build_ldap_sync_plan( + session, + principal, + sync_source, + client=write_client, # type: ignore[arg-type] + ) + else: + raise AddressBookError(f"Sync is not implemented for {sync_source.connector_type} sources.") + _apply_address_sync_plan( session, principal, - sync_source, - force_full=force_full, - password=password, - bearer_token=bearer_token, - client=write_client, + plan, + client=write_client if sync_source.connector_type == "carddav" else None, # type: ignore[arg-type] ) - _apply_address_sync_plan(session, principal, plan, client=write_client) status = "conflict" if plan.stats.conflicts else "succeeded" if plan.stats.errors: status = "failed" @@ -925,6 +1036,275 @@ def run_sync_source( raise +def _build_ldap_sync_plan( + session: Session, + principal: ApiPrincipal, + sync_source: AddressSyncSource, + *, + client: AddressLdapClient | None, +) -> AddressSyncPlan: + settings = _ldap_metadata(sync_source.metadata_) + if not settings: + raise AddressBookError("LDAP sync source configuration is missing.") + ldap_client = client or _ldap_client_for_source(session, sync_source) + attribute_map = { + str(key): str(value) + for key, value in dict(settings.get("attribute_map") or {}).items() + if str(key).strip() and str(value).strip() + } + source_key_attribute = attribute_map.get("source_key") + if not source_key_attribute: + raise AddressBookError("LDAP source mapping requires a stable source_key attribute.") + attributes = tuple( + dict.fromkeys( + [ + *attribute_map.values(), + "entryUUID", + "objectGUID", + "modifyTimestamp", + "uSNChanged", + "entryCSN", + ] + ) + ) + try: + result = ldap_client.search( + base_dn=str(settings.get("base_dn") or sync_source.external_address_book_ref or ""), + search_filter=str(settings.get("search_filter") or "(objectClass=person)"), + attributes=attributes, + page_size=int(settings.get("page_size") or 500), + max_entries=int(settings.get("max_entries") or 10_000), + ) + except AddressLdapError as exc: + raise AddressBookError(str(exc)) from exc + + stats = AddressSyncPlanStats(full_sync=True) + plan = AddressSyncPlan(sync_source=sync_source, stats=stats) + existing = { + str(contact.source_ref): contact + for contact in _ldap_contacts_for_source(session, sync_source) + if contact.source_ref + } + observed_refs: set[str] = set() + revision_rows: list[dict[str, str]] = [] + for entry in result.entries: + serialized_attributes = _json_safe_ldap_attributes(entry.attributes) + source_key = _ldap_scalar(entry.attributes.get(source_key_attribute)) + if not source_key: + source_key = entry.dn.strip() + if not source_key: + _add_sync_plan_item( + plan, + AddressSyncPlanItem( + action="error", + message="LDAP entry has neither the configured source key nor a DN.", + ), + ) + continue + source_ref = f"ldap:{sync_source.id}:{hashlib.sha256(source_key.encode()).hexdigest()}" + if source_ref in observed_refs: + _add_sync_plan_item( + plan, + AddressSyncPlanItem( + action="error", + href=source_ref, + remote_uid=source_key, + message="LDAP search returned a duplicate stable source key.", + ), + ) + continue + observed_refs.add(source_ref) + source_revision = _ldap_source_revision( + entry.attributes, + attribute_map=attribute_map, + serialized_attributes=serialized_attributes, + ) + revision_rows.append({"source_key": source_key, "revision": source_revision}) + try: + payload = _ldap_contact_payload( + entry.attributes, + attribute_map=attribute_map, + sync_source=sync_source, + source_key=source_key, + source_revision=source_revision, + dn=entry.dn, + ) + except (AddressBookError, ValueError) as exc: + _add_sync_plan_item( + plan, + AddressSyncPlanItem( + action="error", + href=source_ref, + remote_uid=source_key, + message=str(exc), + source_revision=source_revision, + ), + ) + continue + local = existing.get(source_ref) + action = "create" + if local is not None: + comparable = payload.model_dump(mode="json", exclude={"provenance"}) + action = ( + "unchanged" + if local.deleted_at is None + and _contact_payload_for_conflict(local) == comparable + and local.source_revision == source_revision + else "update" + ) + _add_sync_plan_item( + plan, + AddressSyncPlanItem( + action=action, + href=source_ref, + remote_uid=source_key, + contact_id=local.id if local is not None else None, + display_name=payload.display_name, + parsed_payload=payload, + source_revision=source_revision, + raw_payload=json.dumps(serialized_attributes, sort_keys=True, ensure_ascii=True), + source_details={"dn": entry.dn, "source_key": source_key}, + ), + ) + + if result.complete: + for source_ref, contact in existing.items(): + if source_ref not in observed_refs and contact.deleted_at is None: + _add_sync_plan_item( + plan, + AddressSyncPlanItem( + action="delete", + href=source_ref, + remote_uid=str((contact.provenance or {}).get("ldap", {}).get("source_key") or "") or None, + contact_id=contact.id, + display_name=contact.display_name, + message="LDAP authoritative source no longer contains this contact.", + ), + ) + else: + _add_sync_plan_item( + plan, + AddressSyncPlanItem( + action="error", + message="LDAP result reached the configured entry limit; absence-based deletes are suppressed.", + ), + ) + stats.remote_revision = hashlib.sha256( + json.dumps(sorted(revision_rows, key=lambda item: item["source_key"]), sort_keys=True).encode() + ).hexdigest() + return plan + + +def _ldap_contact_payload( + attributes: dict[str, Any], + *, + attribute_map: dict[str, str], + sync_source: AddressSyncSource, + source_key: str, + source_revision: str, + dn: str, +) -> ContactCreateRequest: + def scalar(target: str) -> str | None: + attribute = attribute_map.get(target) + return _ldap_scalar(attributes.get(attribute)) if attribute else None + + given_name = scalar("given_name") + family_name = scalar("family_name") + email = scalar("email") + organization = scalar("organization") + display_name = scalar("display_name") or " ".join( + value for value in (given_name, family_name) if value + ) or email or organization + if not display_name: + raise AddressBookError(f'LDAP entry "{dn or source_key}" has no mapped contact identity.') + phone = scalar("phone") + postal = { + target: scalar(target) + for target in ("street", "postal_code", "locality", "region", "country") + } + tag_attribute = attribute_map.get("tags") + tags = [str(item).strip() for item in _ldap_values(attributes.get(tag_attribute)) if str(item).strip()] if tag_attribute else [] + return ContactCreateRequest( + display_name=display_name, + given_name=given_name, + family_name=family_name, + organization=organization, + role_title=scalar("role_title"), + note=scalar("note"), + tags=tags, + emails=[ContactEmailPayload(email=email, is_primary=True)] if email else [], + phones=[ContactPhonePayload(phone=phone, is_primary=True)] if phone else [], + postal_addresses=[ContactPostalAddressPayload(**postal, is_primary=True)] if any(postal.values()) else [], + provenance={ + "ldap": { + "sync_source_id": sync_source.id, + "dn": dn, + "source_key": source_key, + "source_revision": source_revision, + "authority": "external_authoritative", + } + }, + ) + + +def _ldap_source_revision( + attributes: dict[str, Any], + *, + attribute_map: dict[str, str], + serialized_attributes: dict[str, Any], +) -> str: + configured = attribute_map.get("source_revision") + for attribute in (configured, "modifyTimestamp", "uSNChanged", "entryCSN"): + if attribute: + value = _ldap_scalar(attributes.get(attribute)) + if value: + return value + return hashlib.sha256( + json.dumps(serialized_attributes, sort_keys=True, ensure_ascii=True).encode() + ).hexdigest() + + +def _ldap_values(value: Any) -> tuple[Any, ...]: + if value is None: + return () + if isinstance(value, (list, tuple, set)): + return tuple(value) + return (value,) + + +def _ldap_scalar(value: Any) -> str | None: + values = _ldap_values(value) + if not values: + return None + selected = values[0] + if isinstance(selected, bytes): + return selected.hex() + normalized = str(selected).strip() + return normalized or None + + +def _json_safe_ldap_attributes(attributes: dict[str, Any]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in attributes.items(): + values = _ldap_values(value) + normalized = [item.hex() if isinstance(item, bytes) else str(item) for item in values] + result[str(key)] = normalized if isinstance(value, (list, tuple, set)) else (normalized[0] if normalized else None) + return result + + +def _ldap_contacts_for_source(session: Session, sync_source: AddressSyncSource) -> list[Contact]: + return ( + session.query(Contact) + .filter( + Contact.address_book_id == sync_source.address_book_id, + Contact.source_kind.in_(("ldap", "active_directory")), + Contact.source_ref.like(f"ldap:{sync_source.id}:%"), + ) + .order_by(Contact.id.asc()) + .all() + ) + + def _build_carddav_sync_plan( session: Session, principal: ApiPrincipal, @@ -1437,10 +1817,10 @@ def _add_sync_plan_item(plan: AddressSyncPlan, item: AddressSyncPlanItem) -> Non def _apply_address_sync_plan(session: Session, principal: ApiPrincipal, plan: AddressSyncPlan, *, client: AddressCardDAVClient | None = None) -> None: for item in plan.items: if item.action == "create": - contact = _upsert_carddav_contact(session, principal, plan.sync_source, item) + contact = _upsert_remote_contact(session, principal, plan.sync_source, item) item.contact_id = contact.id elif item.action == "update": - contact = _upsert_carddav_contact(session, principal, plan.sync_source, item) + contact = _upsert_remote_contact(session, principal, plan.sync_source, item) item.contact_id = contact.id elif item.action == "delete" and item.contact_id: contact = get_visible_contact(session, principal, item.contact_id, include_deleted=True) @@ -1612,14 +1992,14 @@ def _decrement_sync_plan_stats(plan: AddressSyncPlan, action: str) -> None: plan.stats.deleted = max(0, plan.stats.deleted - 1) -def _upsert_carddav_contact( +def _upsert_remote_contact( session: Session, principal: ApiPrincipal, sync_source: AddressSyncSource, item: AddressSyncPlanItem, ) -> Contact: if item.parsed_payload is None: - raise AddressBookError("Remote vCard payload is missing.") + raise AddressBookError("Mapped remote contact payload is missing.") contact = None if item.contact_id: contact = get_visible_contact(session, principal, item.contact_id, include_deleted=True) @@ -1645,18 +2025,30 @@ def _upsert_carddav_contact( contact.tags = _normalize_tags(item.parsed_payload.tags) contact.source_kind = sync_source.connector_type contact.source_ref = item.href - contact.source_payload_kind = "vcard" - contact.source_payload_raw = item.raw_vcard + contact.source_payload_kind = "vcard" if sync_source.connector_type == "carddav" else "ldap-entry" + contact.source_payload_raw = item.raw_vcard if sync_source.connector_type == "carddav" else None contact.source_revision = item.source_revision - contact.provenance = { - **(item.parsed_payload.provenance or {}), - "carddav": { - "sync_source_id": sync_source.id, - "href": item.href, - "remote_uid": item.remote_uid, - "etag": item.etag, - }, - } + if sync_source.connector_type == "carddav": + source_provenance = { + "carddav": { + "sync_source_id": sync_source.id, + "href": item.href, + "remote_uid": item.remote_uid, + "etag": item.etag, + } + } + else: + source_provenance = { + "ldap": { + "sync_source_id": sync_source.id, + "source_ref": item.href, + "source_key": item.remote_uid, + "source_revision": item.source_revision, + "dn": item.source_details.get("dn"), + "authority": "external_authoritative", + } + } + contact.provenance = {**(item.parsed_payload.provenance or {}), **source_provenance} contact.deleted_at = None contact.updated_by_account_id = _account_id(principal) _replace_emails(contact, item.parsed_payload.emails) @@ -1727,6 +2119,52 @@ def _local_contact_changed_after_last_sync(contact: Contact, sync_source: Addres return bool(contact.updated_at and contact.updated_at > sync_source.last_success_at) +def _ldap_client_from_connection_payload( + session: Session, + principal: ApiPrincipal, + payload: AddressLdapConnectionRequest, +) -> AddressLdapClient: + reusable = _resolve_core_address_credential( + session, + tenant_id=principal.tenant_id, + source_id=None, + credential_ref=payload.credential_ref, + ) + bind_dn = _trim(payload.bind_dn) or _credential_username(reusable) + password = _credential_secret(reusable, auth_type="basic") if reusable is not None else None + return AddressLdapClient( + url=payload.url, + bind_dn=bind_dn, + password=password, + start_tls=payload.start_tls, + connect_timeout=payload.connect_timeout, + receive_timeout=payload.receive_timeout, + ) + + +def _ldap_client_for_source( + session: Session, + sync_source: AddressSyncSource, +) -> AddressLdapClient: + settings = _ldap_metadata(sync_source.metadata_) + reusable = _resolve_core_address_credential( + session, + tenant_id=sync_source.tenant_id or "", + source_id=sync_source.id, + credential_ref=settings.get("credential_ref"), + ) + bind_dn = _trim(str(settings.get("bind_dn") or "")) or _credential_username(reusable) + password = _credential_secret(reusable, auth_type="basic") if reusable is not None else None + return AddressLdapClient( + url=str(settings.get("url") or sync_source.external_account_ref or ""), + bind_dn=bind_dn, + password=password, + start_tls=bool(settings.get("start_tls", True)), + connect_timeout=int(settings.get("connect_timeout") or 10), + receive_timeout=int(settings.get("receive_timeout") or 30), + ) + + def _carddav_client_from_payload( session: Session, principal: ApiPrincipal, @@ -1934,13 +2372,17 @@ def resolve_trusted_deployment_carddav_credential_ref(credential_ref: str) -> st def public_address_sync_metadata(metadata: object) -> dict[str, Any]: payload = copy.deepcopy(metadata) if isinstance(metadata, dict) else {} auth = payload.get("carddav") - if not isinstance(auth, dict): - return payload - had_credential = bool(auth.get("secret_encrypted") or auth.get("credential_ref")) - auth["credential_envelope_id"] = _core_credential_id(auth.get("credential_ref")) - auth.pop("secret_encrypted", None) - auth.pop("credential_ref", None) - auth["has_credential"] = had_credential + if isinstance(auth, dict): + had_credential = bool(auth.get("secret_encrypted") or auth.get("credential_ref")) + auth["credential_envelope_id"] = _core_credential_id(auth.get("credential_ref")) + auth.pop("secret_encrypted", None) + auth.pop("credential_ref", None) + auth["has_credential"] = had_credential + ldap = payload.get("ldap") + if isinstance(ldap, dict): + credential_ref = ldap.pop("credential_ref", None) + ldap["credential_envelope_id"] = _core_credential_id(credential_ref) + ldap["has_credential"] = bool(credential_ref) return payload @@ -1952,6 +2394,39 @@ def _assert_no_caller_carddav_credential_ref(credential_ref: str | None) -> None ) +def _assert_reusable_credential_ref(credential_ref: str | None) -> None: + _assert_no_caller_carddav_credential_ref(credential_ref) + + +def _ldap_metadata(metadata: object) -> dict[str, Any]: + if not isinstance(metadata, dict): + return {} + ldap = metadata.get("ldap") + return ldap if isinstance(ldap, dict) else {} + + +def _assert_api_ldap_metadata_safe(metadata: object) -> None: + ldap = _ldap_metadata(metadata) + forbidden = {"credential_ref", "password", "secret", "bind_password"}.intersection(ldap) + if forbidden: + raise AddressBookError( + "LDAP credential references and secrets are server-managed; use the LDAP source endpoint." + ) + + +def _merge_server_owned_ldap_metadata(existing: object, incoming: dict[str, Any]) -> dict[str, Any]: + merged = copy.deepcopy(incoming) + existing_ldap = _ldap_metadata(existing) + credential_ref = existing_ldap.get("credential_ref") + if credential_ref: + ldap = merged.get("ldap") + if not isinstance(ldap, dict): + ldap = {} + merged["ldap"] = ldap + ldap["credential_ref"] = credential_ref + return merged + + def _carddav_auth_metadata(metadata: object) -> dict[str, Any]: if not isinstance(metadata, dict): return {} diff --git a/tests/test_ldap_sync.py b/tests/test_ldap_sync.py new file mode 100644 index 0000000..85f1c52 --- /dev/null +++ b/tests/test_ldap_sync.py @@ -0,0 +1,172 @@ +from __future__ import annotations + +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from govoplan_addresses.backend.db.models import AddressBook, AddressSyncSource, Contact +from govoplan_addresses.backend.ldap import ( + AddressLdapClient, + AddressLdapEntry, + AddressLdapError, + AddressLdapSearchResult, +) +from govoplan_addresses.backend.ldap_schemas import AddressLdapSourceCreateRequest +from govoplan_addresses.backend.service import ( + create_ldap_sync_source, + preview_sync_source, + run_sync_source, +) +from govoplan_core.db.base import Base + + +class Principal: + account_id = "account-1" + group_ids = frozenset() + + @property + def tenant_id(self) -> str: + return "tenant-1" + + def has(self, scope: str) -> bool: + return scope in { + "addresses:address_book:read", + "addresses:address_book:write", + "addresses:contact:read", + "addresses:contact:write", + "addresses:contact:delete", + "addresses:sync:read", + "addresses:sync:write", + } + + +class FakeLdapClient: + def __init__(self, entries: list[AddressLdapEntry], *, complete: bool = True) -> None: + self.entries = entries + self.complete = complete + + def search(self, *, base_dn: str, search_filter: str, attributes: tuple[str, ...], page_size: int, max_entries: int) -> AddressLdapSearchResult: + del search_filter, attributes, max_entries + return AddressLdapSearchResult( + base_dn=base_dn, + entries=tuple(self.entries), + complete=self.complete, + page_size=page_size, + ) + + +def ldap_entry( + key: str, + *, + revision: str = "20260802090000Z", + organization: str = "Analysis Office", +) -> AddressLdapEntry: + return AddressLdapEntry( + dn=f"uid={key},ou=people,dc=example,dc=test", + attributes={ + "entryUUID": key, + "modifyTimestamp": revision, + "displayName": "Ada Lovelace", + "givenName": "Ada", + "sn": "Lovelace", + "mail": "ada@example.test", + "o": organization, + "memberOf": ["cn=analysts,ou=groups,dc=example,dc=test"], + }, + ) + + +class AddressLdapSyncTests(unittest.TestCase): + def setUp(self) -> None: + engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(engine) + self.session = sessionmaker(bind=engine, expire_on_commit=False)() + self.principal = Principal() + self.book = AddressBook( + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + name="Directory", + source_kind="local", + read_only=False, + ) + self.session.add(self.book) + self.session.flush() + self.source = create_ldap_sync_source( + self.session, + self.principal, + self.book.id, + AddressLdapSourceCreateRequest( + url="ldaps://directory.example.test", + display_name="Corporate directory", + base_dn="ou=people,dc=example,dc=test", + ), + ) + self.session.flush() + + def test_preview_is_nonmutating_and_full_sync_is_idempotent(self) -> None: + client = FakeLdapClient([ldap_entry("person-1")]) + preview = preview_sync_source(self.session, self.principal, self.source.id, client=client) + self.assertEqual(1, preview.stats.created) + self.assertEqual(0, self.session.query(Contact).count()) + + first = run_sync_source(self.session, self.principal, self.source.id, client=client) + self.assertEqual(1, first.stats.created) + contact = self.session.query(Contact).one() + self.assertEqual("ldap", contact.source_kind) + self.assertEqual("person-1", contact.provenance["ldap"]["source_key"]) + self.assertEqual("succeeded", self.source.status) + + repeated = run_sync_source(self.session, self.principal, self.source.id, client=client) + self.assertEqual(1, repeated.stats.unchanged) + self.assertEqual(1, self.session.query(Contact).count()) + + changed_client = FakeLdapClient( + [ldap_entry("person-1", revision="20260802100000Z", organization="Computing Office")] + ) + changed = run_sync_source(self.session, self.principal, self.source.id, client=changed_client) + self.assertEqual(1, changed.stats.updated) + self.assertEqual("Computing Office", self.session.query(Contact).one().organization) + + def test_only_complete_scans_plan_authoritative_deletes(self) -> None: + run_sync_source( + self.session, + self.principal, + self.source.id, + client=FakeLdapClient([ldap_entry("person-1")]), + ) + incomplete = preview_sync_source( + self.session, + self.principal, + self.source.id, + client=FakeLdapClient([], complete=False), + ) + self.assertEqual(0, incomplete.stats.deleted) + self.assertEqual(1, incomplete.stats.errors) + self.assertIsNone(self.session.query(Contact).one().deleted_at) + + complete = run_sync_source( + self.session, + self.principal, + self.source.id, + client=FakeLdapClient([], complete=True), + ) + self.assertEqual(1, complete.stats.deleted) + self.assertIsNotNone(self.session.query(Contact).one().deleted_at) + + def test_connector_requires_encrypted_transport(self) -> None: + with self.assertRaisesRegex(AddressLdapError, "require StartTLS"): + AddressLdapClient(url="ldap://directory.example.test", start_tls=False) + with self.assertRaisesRegex(AddressLdapError, "must not contain credentials"): + AddressLdapClient(url="ldaps://user:secret@directory.example.test") + + def test_source_is_always_read_only(self) -> None: + source = self.session.get(AddressSyncSource, self.source.id) + self.assertTrue(source.read_only) + self.assertEqual("read_only", source.sync_direction) + self.assertTrue(source.address_book.read_only) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..8478fb0 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from alembic.runtime.migration import MigrationContext +from sqlalchemy import create_engine, inspect + +from govoplan_addresses.backend.manifest import get_manifest +from govoplan_core.db.migrations import migrate_database + + +class AddressesMigrationTests(unittest.TestCase): + def test_fresh_database_reaches_import_profile_head(self) -> None: + with tempfile.TemporaryDirectory(prefix="govoplan-addresses-migration-") as directory: + url = f"sqlite:///{Path(directory) / 'addresses.db'}" + migrate_database( + database_url=url, + enabled_modules=("addresses",), + manifest_factories=(get_manifest,), + ) + engine = create_engine(url) + try: + with engine.connect() as connection: + self.assertIn( + "c5d7e8f9a0b1", + set(MigrationContext.configure(connection).get_current_heads()), + ) + tables = set(inspect(connection).get_table_names()) + self.assertIn("addresses_import_profiles", tables) + self.assertIn("addresses_import_runs", tables) + finally: + engine.dispose() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_provider_state.py b/tests/test_provider_state.py index 4c0a9ea..fb10b91 100644 --- a/tests/test_provider_state.py +++ b/tests/test_provider_state.py @@ -15,7 +15,9 @@ from govoplan_addresses.backend.db.models import ( from govoplan_addresses.backend.manifest import manifest from govoplan_addresses.backend.provider_state import ( CARDDAV_PROVIDER_ID, + LDAP_PROVIDER_ID, carddav_provider_states, + ldap_provider_states, ) from govoplan_core.core.provider_governance import ExternalProviderStateContext from govoplan_core.db.base import Base @@ -103,6 +105,28 @@ class AddressesProviderStateTests(unittest.TestCase): CARDDAV_PROVIDER_ID, manifest.external_provider_state_providers[0].provider_id, ) + self.assertEqual(LDAP_PROVIDER_ID, manifest.external_providers[1].id) + self.assertEqual( + LDAP_PROVIDER_ID, + manifest.external_provider_state_providers[1].provider_id, + ) + + def test_failed_ldap_source_is_stale_without_exposing_endpoint(self) -> None: + self.source.connector_type = "ldap" + self.source.display_name = "Directory" + self.source.status = "failed" + self.source.last_error = "connection failed" + self.session.flush() + + state = ldap_provider_states( + ExternalProviderStateContext(session=self.session, tenant_id="tenant-1") + )[0] + + self.assertEqual(LDAP_PROVIDER_ID, state.provider_id) + self.assertEqual("error", state.health) + self.assertEqual("current", state.freshness) + self.assertEqual("external_authoritative", state.authority_mode) + self.assertNotIn("dav.example.test", str(state.to_dict())) if __name__ == "__main__": diff --git a/tests/test_tabular_imports.py b/tests/test_tabular_imports.py new file mode 100644 index 0000000..d5096c0 --- /dev/null +++ b/tests/test_tabular_imports.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import base64 +from io import BytesIO +import unittest + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from openpyxl import Workbook + +from govoplan_addresses.backend.db.models import AddressBook, Contact +from govoplan_addresses.backend.import_schemas import ( + AddressImportConfiguration, + AddressImportPreviewRequest, + AddressImportProfileCreateRequest, + AddressImportProfileUpdateRequest, + AddressImportRollbackRequest, +) +from govoplan_addresses.backend.imports import ( + apply_address_import, + create_import_profile, + import_run_payload, + preview_address_import, + rollback_address_import, + update_import_profile, +) +from govoplan_core.db.base import Base + + +class Principal: + account_id = "account-1" + group_ids = frozenset({"group-1"}) + + @property + def tenant_id(self) -> str: + return "tenant-1" + + def has(self, scope: str) -> bool: + return scope in { + "addresses:address_book:read", + "addresses:address_book:write", + "addresses:contact:read", + "addresses:contact:write", + } + + +def encoded(value: str) -> str: + return base64.b64encode(value.encode()).decode() + + +class AddressTabularImportTests(unittest.TestCase): + def setUp(self) -> None: + engine = create_engine("sqlite+pysqlite:///:memory:") + Base.metadata.create_all(engine) + self.session = sessionmaker(bind=engine, expire_on_commit=False)() + self.principal = Principal() + self.book = AddressBook( + tenant_id="tenant-1", + scope_type="tenant", + scope_id="tenant-1", + name="Imported contacts", + source_kind="local", + read_only=False, + ) + self.session.add(self.book) + self.profile = create_import_profile( + self.session, + self.principal, + AddressImportProfileCreateRequest( + scope_type="tenant", + name="Monthly contacts", + source_format="csv", + configuration=AddressImportConfiguration( + delimiter=";", + field_mappings={ + "source_key": "id", + "given_name": "first", + "family_name": "last", + "email": "email", + "organization": "organization", + }, + ), + ), + ) + self.session.flush() + + def test_preview_apply_repeat_and_guarded_rollback(self) -> None: + payload = AddressImportPreviewRequest( + profile_id=self.profile.id, + filename="contacts.csv", + content_base64=encoded( + "id;first;last;email;organization\n" + "1;Ada;Lovelace;ada@example.test;Analysis Office\n" + "2;Grace;Hopper;grace@example.test;Computing Office\n" + ), + ) + run = preview_address_import(self.session, self.principal, self.book.id, payload) + self.assertEqual(2, run.statistics["create"]) + self.assertFalse(run.diagnostics) + + applied = apply_address_import( + self.session, + self.principal, + run.id, + expected_plan_hash=run.plan_hash, + ) + self.assertEqual("applied", applied.status) + self.assertEqual(2, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count()) + self.assertIs(applied, apply_address_import(self.session, self.principal, run.id, expected_plan_hash=run.plan_hash)) + response_evidence = import_run_payload(applied)["result_evidence"] + self.assertEqual(2, response_evidence["created_contact_count"]) + self.assertNotIn("created_contact_ids", response_evidence) + self.assertNotIn("updated_contacts", response_evidence) + + repeated = preview_address_import(self.session, self.principal, self.book.id, payload) + self.assertEqual(2, repeated.statistics["unchanged"]) + apply_address_import(self.session, self.principal, repeated.id, expected_plan_hash=repeated.plan_hash) + self.assertEqual(2, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count()) + + rolled_back = rollback_address_import( + self.session, + self.principal, + run.id, + AddressImportRollbackRequest(reason="The operator selected the wrong monthly file."), + ) + self.assertEqual("rolled_back", rolled_back.status) + self.assertEqual(0, self.session.query(Contact).filter(Contact.deleted_at.is_(None)).count()) + + def test_duplicate_keys_and_changed_targets_block_apply(self) -> None: + duplicate = preview_address_import( + self.session, + self.principal, + self.book.id, + AddressImportPreviewRequest( + profile_id=self.profile.id, + filename="duplicates.csv", + content_base64=encoded( + "id;first;last;email;organization\n" + "1;Ada;Lovelace;ada@example.test;One\n" + "1;Ada;Lovelace;ada@example.test;Two\n" + ), + ), + ) + self.assertEqual(2, duplicate.statistics["conflict"]) + with self.assertRaisesRegex(ValueError, "error diagnostics"): + apply_address_import(self.session, self.principal, duplicate.id, expected_plan_hash=duplicate.plan_hash) + + initial = preview_address_import( + self.session, + self.principal, + self.book.id, + AddressImportPreviewRequest( + profile_id=self.profile.id, + filename="one.csv", + content_base64=encoded("id;first;last;email;organization\n1;Ada;Lovelace;ada@example.test;One\n"), + ), + ) + apply_address_import(self.session, self.principal, initial.id, expected_plan_hash=initial.plan_hash) + changed = preview_address_import( + self.session, + self.principal, + self.book.id, + AddressImportPreviewRequest( + profile_id=self.profile.id, + filename="one.csv", + content_base64=encoded("id;first;last;email;organization\n1;Ada;Lovelace;ada@example.test;Two\n"), + ), + ) + contact = self.session.query(Contact).filter(Contact.deleted_at.is_(None)).one() + contact.organization = "Concurrent edit" + self.session.flush() + with self.assertRaisesRegex(ValueError, "changed after preview"): + apply_address_import(self.session, self.principal, changed.id, expected_plan_hash=changed.plan_hash) + + def test_profile_updates_create_immutable_versions(self) -> None: + next_profile = update_import_profile( + self.session, + self.principal, + self.profile.id, + payload=AddressImportProfileUpdateRequest(name="Monthly contacts v2"), + ) + self.assertFalse(self.profile.is_current) + self.assertTrue(next_profile.is_current) + self.assertEqual(self.profile.profile_key, next_profile.profile_key) + self.assertEqual(2, next_profile.version) + + def test_xlsx_sheet_selection_and_formula_rejection(self) -> None: + workbook = Workbook() + workbook.active.title = "Ignore" + sheet = workbook.create_sheet("Contacts") + sheet.append(["id", "first", "last", "email", "organization"]) + sheet.append(["1", "Ada", "Lovelace", "ada@example.test", "Analysis Office"]) + content = BytesIO() + workbook.save(content) + xlsx_profile = create_import_profile( + self.session, + self.principal, + AddressImportProfileCreateRequest( + scope_type="tenant", + name="Workbook contacts", + source_format="xlsx", + configuration=AddressImportConfiguration( + sheet_name="Contacts", + field_mappings={ + "source_key": "id", + "given_name": "first", + "family_name": "last", + "email": "email", + "organization": "organization", + }, + ), + ), + ) + self.session.flush() + run = preview_address_import( + self.session, + self.principal, + self.book.id, + AddressImportPreviewRequest( + profile_id=xlsx_profile.id, + filename="contacts.xlsx", + content_base64=base64.b64encode(content.getvalue()).decode(), + ), + ) + self.assertEqual(1, run.statistics["create"]) + + sheet["E2"] = "=CONCAT(\"Analysis\", \" Office\")" + content = BytesIO() + workbook.save(content) + with self.assertRaisesRegex(ValueError, "formulas are never evaluated"): + preview_address_import( + self.session, + self.principal, + self.book.id, + AddressImportPreviewRequest( + profile_id=xlsx_profile.id, + filename="contacts.xlsx", + content_base64=base64.b64encode(content.getvalue()).decode(), + ), + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/src/api/addresses.ts b/webui/src/api/addresses.ts index 9d3803e..1af8d38 100644 --- a/webui/src/api/addresses.ts +++ b/webui/src/api/addresses.ts @@ -484,6 +484,92 @@ export type AddressSyncConflict = { updated_at: string; }; +export type AddressImportConfiguration = { + field_mappings: Record; + delimiter: "," | ";" | "\t" | "|"; + encoding: "utf-8" | "utf-8-sig" | "cp1252" | "latin-1"; + header_row: number; + sheet_name?: string | null; + source_key_column?: string | null; + duplicate_source_key_policy: "reject" | "first" | "last"; + existing_contact_policy: "update" | "ignore" | "reject"; + blank_value_policy: "ignore" | "clear" | "reject"; + locale?: string | null; + default_tags: string[]; + max_rows: number; +}; + +export type AddressImportProfile = { + id: string; + profile_key: string; + version: number; + tenant_id?: string | null; + scope_type: AddressBookScope; + scope_id?: string | null; + name: string; + description?: string | null; + source_format: "csv" | "xlsx"; + configuration: AddressImportConfiguration; + is_current: boolean; + created_at: string; + updated_at: string; +}; + +export type AddressImportEffect = { + row_number: number; + action: "create" | "update" | "conflict" | "unchanged" | "ignored"; + source_key?: string | null; + contact_id?: string | null; + display_name?: string | null; + changed_fields: string[]; + message?: string | null; +}; + +export type AddressImportDiagnostic = { + severity: "info" | "warning" | "error"; + code: string; + message: string; + row_number?: number | null; + field?: string | null; + details: Record; +}; + +export type AddressImportRun = { + id: string; + address_book_id: string; + profile_id: string; + source_filename: string; + source_format: string; + input_hash: string; + plan_hash: string; + status: string; + row_count: number; + statistics: Record; + diagnostics: AddressImportDiagnostic[]; + effects: AddressImportEffect[]; + can_apply: boolean; + result_evidence: Record; + created_at: string; + updated_at: string; + applied_at?: string | null; + rolled_back_at?: string | null; +}; + +export type AddressLdapSourcePayload = { + url: string; + credential_ref?: string | null; + bind_dn?: string | null; + start_tls: boolean; + connect_timeout?: number; + receive_timeout?: number; + display_name: string; + base_dn: string; + search_filter: string; + page_size: number; + max_entries: number; + attribute_map: Record; +}; + type AddressSyncSourceListResponse = { sync_sources: AddressSyncSource[]; }; @@ -508,6 +594,14 @@ type AddressSyncConflictListResponse = { conflicts: AddressSyncConflict[]; }; +type AddressImportProfileListResponse = { + profiles: AddressImportProfile[]; +}; + +type AddressLdapDiscoveryResponse = { + base_dns: string[]; +}; + type ContactChannelRuleListResponse = { rules: ContactChannelRule[]; }; @@ -692,6 +786,27 @@ export function createCardDavSyncSource( }); } +export function discoverLdapBaseDns( + settings: ApiSettings, + payload: Pick +): Promise { + return apiFetch(settings, "/api/v1/addresses/ldap/discover", { + method: "POST", + body: JSON.stringify(payload) + }).then((response) => response.base_dns); +} + +export function createLdapSyncSource( + settings: ApiSettings, + addressBookId: string, + payload: AddressLdapSourcePayload +): Promise { + return apiFetch(settings, `/api/v1/addresses/address-books/${addressBookId}/ldap/sources`, { + method: "POST", + body: JSON.stringify(payload) + }); +} + export async function listAddressCredentials( settings: ApiSettings, sourceId?: string | null @@ -941,6 +1056,64 @@ export function importAddressBookVcards(settings: ApiSettings, addressBookId: st }); } +export async function listAddressImportProfiles(settings: ApiSettings): Promise { + const response = await apiFetch(settings, "/api/v1/addresses/import-profiles"); + return response.profiles; +} + +export function createAddressImportProfile( + settings: ApiSettings, + payload: { + scope_type: AddressBookScope; + scope_id?: string | null; + name: string; + description?: string | null; + source_format: "csv" | "xlsx"; + configuration: AddressImportConfiguration; + } +): Promise { + return apiFetch(settings, "/api/v1/addresses/import-profiles", { + method: "POST", + body: JSON.stringify(payload) + }); +} + +export function updateAddressImportProfile( + settings: ApiSettings, + profileId: string, + payload: { name?: string; description?: string | null; configuration?: AddressImportConfiguration } +): Promise { + return apiFetch(settings, `/api/v1/addresses/import-profiles/${profileId}`, { + method: "PATCH", + body: JSON.stringify(payload) + }); +} + +export function previewAddressImport( + settings: ApiSettings, + addressBookId: string, + payload: { profile_id: string; filename: string; content_base64: string } +): Promise { + return apiFetch(settings, `/api/v1/addresses/address-books/${addressBookId}/imports/preview`, { + method: "POST", + body: JSON.stringify(payload) + }); +} + +export function applyAddressImport(settings: ApiSettings, run: AddressImportRun): Promise { + return apiFetch(settings, `/api/v1/addresses/imports/${run.id}/apply`, { + method: "POST", + body: JSON.stringify({ expected_plan_hash: run.plan_hash }) + }); +} + +export function rollbackAddressImport(settings: ApiSettings, runId: string, reason: string): Promise { + return apiFetch(settings, `/api/v1/addresses/imports/${runId}/rollback`, { + method: "POST", + body: JSON.stringify({ reason }) + }); +} + export function exportAddressBookVcards(settings: ApiSettings, addressBookId: string): Promise { return apiFetch(settings, `/api/v1/addresses/address-books/${addressBookId}/vcards/export`); } diff --git a/webui/src/features/addressbook/AddressBookPage.tsx b/webui/src/features/addressbook/AddressBookPage.tsx index 708f260..b73e7ea 100644 --- a/webui/src/features/addressbook/AddressBookPage.tsx +++ b/webui/src/features/addressbook/AddressBookPage.tsx @@ -1,4 +1,4 @@ -import { Download, Edit3, GitMerge, History, Link2, Plus, RefreshCw, RotateCcw, Save, Search, ShieldCheck, Trash2, Upload, UserPlus, X } from "lucide-react"; +import { Download, Edit3, GitMerge, History, Link2, Network, Plus, RefreshCw, RotateCcw, Save, Search, ShieldCheck, Trash2, Upload, UserPlus, X } from "lucide-react"; import { useCallback, useEffect, useMemo, useState, type DragEvent as ReactDragEvent, type FormEvent } from "react"; import { ApiError, @@ -29,7 +29,9 @@ import { createAddressBook, createAddressList, createAddressListEntry, + createAddressImportProfile, createCardDavSyncSource, + createLdapSyncSource, createContact, createContactChannelRule, createContactQualityDecision, @@ -39,12 +41,15 @@ import { deleteContact, deleteAddressSyncSource, discoverCardDavAddressBooks, + discoverLdapBaseDns, endContactChannelRule, exportAddressBookVcards, exportContactVcard, importAddressBookVcards, + applyAddressImport, getAddressQualitySummary, listAddressBooks, + listAddressImportProfiles, listAddressCredentials, listAddressListEntries, listAddressLists, @@ -59,6 +64,7 @@ import { listContactMerges, listContactProvenance, previewAddressSyncSource, + previewAddressImport, mergeContacts, recoverContactMerge, restoreAddressBook, @@ -67,10 +73,13 @@ import { resolveAddressSyncConflict, runAddressSyncSource, updateAddressBook, + updateAddressImportProfile, updateAddressList, updateAddressSyncSource, updateContact, type AddressCardDavAddressBook, + type AddressImportProfile, + type AddressImportRun, type AddressBook, type AddressBookScope, type AddressChannelDecision, @@ -190,6 +199,37 @@ type CardDavFormState = { sync_direction: "read_only" | "import" | "export" | "two_way"; }; +type LdapFormState = { + url: string; + display_name: string; + credential_envelope_id: string; + bind_dn: string; + start_tls: boolean; + base_dn: string; + search_filter: string; + page_size: string; + max_entries: string; + attribute_map: Record; +}; + +type ImportMode = "vcard" | "tabular"; + +type ImportProfileFormState = { + name: string; + source_format: "csv" | "xlsx"; + delimiter: "," | ";" | "\t" | "|"; + encoding: "utf-8" | "utf-8-sig" | "cp1252" | "latin-1"; + header_row: string; + sheet_name: string; + duplicate_source_key_policy: "reject" | "first" | "last"; + existing_contact_policy: "update" | "ignore" | "reject"; + blank_value_policy: "ignore" | "clear" | "reject"; + locale: string; + default_tags: string; + max_rows: string; + field_mappings: Record; +}; + type SyncInspectorState = { source: AddressSyncSource; } | null; @@ -306,6 +346,90 @@ const EMPTY_CARDDAV_FORM: CardDavFormState = { sync_direction: "read_only" }; +const DEFAULT_LDAP_ATTRIBUTE_MAP: Record = { + source_key: "entryUUID", + source_revision: "modifyTimestamp", + display_name: "displayName", + given_name: "givenName", + family_name: "sn", + organization: "o", + role_title: "title", + email: "mail", + phone: "telephoneNumber", + street: "streetAddress", + postal_code: "postalCode", + locality: "l", + region: "st", + country: "c", + tags: "memberOf" +}; + +const EMPTY_LDAP_FORM: LdapFormState = { + url: "ldaps://", + display_name: "", + credential_envelope_id: "", + bind_dn: "", + start_tls: true, + base_dn: "", + search_filter: "(&(objectClass=person)(mail=*))", + page_size: "500", + max_entries: "10000", + attribute_map: DEFAULT_LDAP_ATTRIBUTE_MAP +}; + +const EMPTY_IMPORT_PROFILE_FORM: ImportProfileFormState = { + name: "", + source_format: "csv", + delimiter: ";", + encoding: "utf-8-sig", + header_row: "1", + sheet_name: "", + duplicate_source_key_policy: "reject", + existing_contact_policy: "update", + blank_value_policy: "ignore", + locale: "", + default_tags: "", + max_rows: "10000", + field_mappings: { + source_key: "id", + display_name: "display_name", + given_name: "given_name", + family_name: "family_name", + organization: "organization", + role_title: "role_title", + email: "email", + phone: "phone", + street: "street", + postal_code: "postal_code", + locality: "locality", + region: "region", + country: "country", + tags: "tags" + } +}; + +const IMPORT_MAPPING_FIELDS = [ + ["source_key", "Stable source key"], + ["display_name", "Display name"], + ["given_name", "Given name"], + ["family_name", "Family name"], + ["organization", "Organization"], + ["role_title", "Role title"], + ["email", "Email"], + ["phone", "Phone"], + ["street", "Street"], + ["postal_code", "Postal code"], + ["locality", "Locality"], + ["region", "Region"], + ["country", "Country"], + ["tags", "Tags"] +] as const; + +const LDAP_MAPPING_FIELDS = [ + ["source_revision", "Source revision"], + ...IMPORT_MAPPING_FIELDS +] as const; + const EMPTY_CHANNEL_RULE_FORM: ChannelRuleFormState = { channel: "email", purpose: "", @@ -756,6 +880,20 @@ function downloadText(filename: string, content: string, type = "text/vcard;char window.URL.revokeObjectURL(url); } +function fileAsBase64(file: File): Promise { + return new Promise((resolve, reject) => { + const reader = new FileReader(); + reader.onerror = () => reject(reader.error ?? new Error("The selected file could not be read.")); + reader.onload = () => { + const value = String(reader.result ?? ""); + const separator = value.indexOf(","); + if (separator < 0) reject(new Error("The selected file did not produce readable content.")); + else resolve(value.slice(separator + 1)); + }; + reader.readAsDataURL(file); + }); +} + function channelRuleState(rule: ContactChannelRule): "active" | "scheduled" | "ended" { const now = Date.now(); if (rule.effective_until && new Date(rule.effective_until).getTime() <= now) return "ended"; @@ -814,12 +952,23 @@ export default function AddressBookPage({ settings, auth, onAuthChange }: Props) const [memberTargetByContactId, setMemberTargetByContactId] = useState>({}); const [dropTargetListId, setDropTargetListId] = useState(""); const [importOpen, setImportOpen] = useState(false); + const [importMode, setImportMode] = useState("vcard"); const [vcardContent, setVcardContent] = useState(""); + const [importProfiles, setImportProfiles] = useState([]); + const [selectedImportProfileId, setSelectedImportProfileId] = useState(""); + const [creatingImportProfile, setCreatingImportProfile] = useState(false); + const [editingImportProfileId, setEditingImportProfileId] = useState(""); + const [importProfileForm, setImportProfileForm] = useState(EMPTY_IMPORT_PROFILE_FORM); + const [importFile, setImportFile] = useState(null); + const [importRun, setImportRun] = useState(null); const [cardDavOpen, setCardDavOpen] = useState(false); const [cardDavForm, setCardDavForm] = useState(EMPTY_CARDDAV_FORM); const [cardDavDiscovery, setCardDavDiscovery] = useState([]); const [cardDavCredentials, setCardDavCredentials] = useState([]); const [cardDavCredentialsError, setCardDavCredentialsError] = useState(""); + const [ldapOpen, setLdapOpen] = useState(false); + const [ldapForm, setLdapForm] = useState(EMPTY_LDAP_FORM); + const [ldapBaseDns, setLdapBaseDns] = useState([]); const [syncInspector, setSyncInspector] = useState(null); const [syncPlan, setSyncPlan] = useState(null); const [syncDiagnostics, setSyncDiagnostics] = useState([]); @@ -842,7 +991,7 @@ export default function AddressBookPage({ settings, auth, onAuthChange }: Props) const canWriteSync = hasScope(auth, "addresses:sync:write"); useEffect(() => { - if (!cardDavOpen || !canWriteSync) { + if ((!cardDavOpen && !ldapOpen) || !canWriteSync) { setCardDavCredentials([]); setCardDavCredentialsError(""); return; @@ -867,11 +1016,28 @@ export default function AddressBookPage({ settings, auth, onAuthChange }: Props) }, [ canWriteSync, cardDavOpen, + ldapOpen, settings.accessToken, settings.apiBaseUrl, settings.apiKey ]); + useEffect(() => { + if (!importOpen || importMode !== "tabular") return; + let active = true; + listAddressImportProfiles(settings) + .then((profiles) => { + if (!active) return; + setImportProfiles(profiles); + setSelectedImportProfileId((current) => current || profiles[0]?.id || ""); + setCreatingImportProfile(profiles.length === 0); + }) + .catch((err) => { + if (active) setError(errorMessage(err)); + }); + return () => { active = false; }; + }, [importMode, importOpen, settings.accessToken, settings.apiBaseUrl, settings.apiKey]); + useEffect(() => { if (auth.groups_loaded || !onAuthChange) return; let active = true; @@ -1033,6 +1199,29 @@ export default function AddressBookPage({ settings, auth, onAuthChange }: Props) [!selectedBook, "Select an address book before importing vCards."], [!vcardContent.trim(), "Paste vCard content before importing."] ); + const importProfileSaveReason = disabledReason( + [saving, savingReason], + [!importProfileForm.name.trim(), "Enter a mapping profile name."], + [!importProfileForm.field_mappings.source_key?.trim(), "Map a stable source-key column."] + ); + const tabularPreviewReason = disabledReason( + [saving, savingReason], + [!selectedBook, "Select an address book before importing."], + [!selectedImportProfileId, "Select or create a mapping profile."], + [!importFile, "Select a CSV or XLSX file."] + ); + const tabularApplyReason = disabledReason( + [saving, savingReason], + [!importRun, "Preview the import first."], + [!importRun?.can_apply, "Resolve all import diagnostics before applying."], + [importRun?.status !== "previewed", "This import plan is no longer pending."] + ); + const connectLdapReason = disabledReason( + [!selectedBook, "Select an address book before connecting LDAP."], + [!canWriteSync, "You need permission to manage address sync."], + [Boolean(selectedBook?.deleted_at), "Restore this address book before connecting sync."], + [saving, savingReason] + ); const addContactRowReason = disabledReason([saving, savingReason]); const removeEmailRowReason = disabledReason( [saving, savingReason], @@ -1990,6 +2179,154 @@ export default function AddressBookPage({ settings, auth, onAuthChange }: Props) } } + function openImportDialog() { + setImportMode("vcard"); + setImportRun(null); + setImportFile(null); + setCreatingImportProfile(false); + setEditingImportProfileId(""); + setImportProfileForm(EMPTY_IMPORT_PROFILE_FORM); + setImportOpen(true); + } + + async function saveImportProfile() { + if (!selectedBook) return; + setSaving(true); + setError(""); + try { + const fieldMappings = Object.fromEntries( + Object.entries(importProfileForm.field_mappings) + .map(([key, value]) => [key, value.trim()]) + .filter(([, value]) => Boolean(value)) + ); + const configuration = { + field_mappings: fieldMappings, + delimiter: importProfileForm.delimiter, + encoding: importProfileForm.encoding, + header_row: Number(importProfileForm.header_row) || 1, + sheet_name: importProfileForm.source_format === "xlsx" ? importProfileForm.sheet_name.trim() || null : null, + source_key_column: null, + duplicate_source_key_policy: importProfileForm.duplicate_source_key_policy, + existing_contact_policy: importProfileForm.existing_contact_policy, + blank_value_policy: importProfileForm.blank_value_policy, + locale: importProfileForm.locale.trim() || null, + default_tags: importProfileForm.default_tags.split(",").map((tag) => tag.trim()).filter(Boolean), + max_rows: Number(importProfileForm.max_rows) || 10000 + }; + const profile = editingImportProfileId + ? await updateAddressImportProfile(settings, editingImportProfileId, { + name: importProfileForm.name.trim(), + configuration + }) + : await createAddressImportProfile(settings, { + scope_type: selectedBook.scope_type, + scope_id: selectedBook.scope_id ?? null, + name: importProfileForm.name.trim(), + source_format: importProfileForm.source_format, + configuration + }); + setImportProfiles((current) => [ + ...current.filter((item) => item.profile_key !== profile.profile_key), + profile + ].sort((left, right) => left.name.localeCompare(right.name))); + setSelectedImportProfileId(profile.id); + setCreatingImportProfile(false); + setEditingImportProfileId(""); + setImportRun(null); + setNotice(`Saved import mapping "${profile.name}" version ${profile.version}.`); + } catch (err) { + setError(errorMessage(err)); + } finally { + setSaving(false); + } + } + + function startNewImportProfile() { + setEditingImportProfileId(""); + setImportProfileForm(EMPTY_IMPORT_PROFILE_FORM); + setCreatingImportProfile(true); + setImportRun(null); + } + + function editSelectedImportProfile() { + const profile = importProfiles.find((item) => item.id === selectedImportProfileId); + if (!profile) return; + const config = profile.configuration; + setEditingImportProfileId(profile.id); + setImportProfileForm({ + name: profile.name, + source_format: profile.source_format, + delimiter: config.delimiter, + encoding: config.encoding, + header_row: String(config.header_row), + sheet_name: config.sheet_name ?? "", + duplicate_source_key_policy: config.duplicate_source_key_policy, + existing_contact_policy: config.existing_contact_policy, + blank_value_policy: config.blank_value_policy, + locale: config.locale ?? "", + default_tags: config.default_tags.join(", "), + max_rows: String(config.max_rows), + field_mappings: { ...config.field_mappings } + }); + setCreatingImportProfile(true); + setImportRun(null); + } + + function cancelImportProfileEditor() { + setCreatingImportProfile(false); + setEditingImportProfileId(""); + setImportProfileForm(EMPTY_IMPORT_PROFILE_FORM); + } + + function downloadImportCorrections() { + if (!importRun) return; + const quote = (value: unknown) => `"${String(value ?? "").replaceAll('"', '""')}"`; + const lines = [ + ["severity", "row", "field", "code", "message"].map(quote).join(","), + ...importRun.diagnostics.map((item) => [item.severity, item.row_number ?? "", item.field ?? "", item.code, item.message].map(quote).join(",")) + ]; + downloadText(`${importRun.source_filename}.corrections.csv`, lines.join("\r\n"), "text/csv;charset=utf-8"); + } + + async function previewTabularImport() { + if (!selectedBook || !importFile || !selectedImportProfileId) return; + setSaving(true); + setError(""); + setNotice(""); + try { + const content = await fileAsBase64(importFile); + const run = await previewAddressImport(settings, selectedBook.id, { + profile_id: selectedImportProfileId, + filename: importFile.name, + content_base64: content + }); + setImportRun(run); + setNotice(`Previewed ${run.row_count} row${run.row_count === 1 ? "" : "s"}.`); + } catch (err) { + setError(errorMessage(err)); + } finally { + setSaving(false); + } + } + + async function applyTabularImport() { + if (!selectedBook || !importRun) return; + setSaving(true); + setError(""); + setNotice(""); + try { + const run = await applyAddressImport(settings, importRun); + setImportRun(run); + setNotice(`Applied import plan: ${run.statistics.create ?? 0} created, ${run.statistics.update ?? 0} updated.`); + await refreshBooks(); + await refreshContacts(selectedBook.id, query); + } catch (err) { + setError(errorMessage(err)); + } finally { + setSaving(false); + } + } + function openCardDavDialog() { setCardDavForm(EMPTY_CARDDAV_FORM); setCardDavDiscovery([]); @@ -2065,6 +2402,72 @@ export default function AddressBookPage({ settings, auth, onAuthChange }: Props) } } + function openLdapDialog() { + setLdapForm({ ...EMPTY_LDAP_FORM, attribute_map: { ...DEFAULT_LDAP_ATTRIBUTE_MAP } }); + setLdapBaseDns([]); + setCardDavCredentialsError(""); + setLdapOpen(true); + } + + function ldapConnectionPayload() { + return { + url: ldapForm.url.trim(), + credential_ref: ldapForm.credential_envelope_id ? `credential-envelope:${ldapForm.credential_envelope_id}` : null, + bind_dn: ldapForm.bind_dn.trim() || null, + start_tls: ldapForm.start_tls, + connect_timeout: 10, + receive_timeout: 30 + }; + } + + async function discoverLdapSources() { + setSaving(true); + setError(""); + setNotice(""); + try { + const baseDns = await discoverLdapBaseDns(settings, ldapConnectionPayload()); + setLdapBaseDns(baseDns); + if (!ldapForm.base_dn && baseDns[0]) setLdapForm((current) => ({ ...current, base_dn: baseDns[0] })); + setNotice(`Found ${baseDns.length} LDAP base DN${baseDns.length === 1 ? "" : "s"}.`); + } catch (err) { + setError(errorMessage(err)); + } finally { + setSaving(false); + } + } + + async function submitLdapSource(event: FormEvent) { + event.preventDefault(); + if (!selectedBook) return; + setSaving(true); + setError(""); + setNotice(""); + try { + const attributeMap = Object.fromEntries( + Object.entries(ldapForm.attribute_map) + .map(([key, value]) => [key, value.trim()]) + .filter(([, value]) => Boolean(value)) + ); + await createLdapSyncSource(settings, selectedBook.id, { + ...ldapConnectionPayload(), + display_name: ldapForm.display_name.trim() || "LDAP directory", + base_dn: ldapForm.base_dn.trim(), + search_filter: ldapForm.search_filter.trim(), + page_size: Number(ldapForm.page_size) || 500, + max_entries: Number(ldapForm.max_entries) || 10000, + attribute_map: attributeMap + }); + setLdapOpen(false); + setLdapBaseDns([]); + setNotice("LDAP / Active Directory source connected."); + await refreshBooks(); + } catch (err) { + setError(errorMessage(err)); + } finally { + setSaving(false); + } + } + async function loadSyncDetails(source: AddressSyncSource) { const [diagnostics, tombstones, conflicts] = await Promise.all([ listAddressSyncDiagnostics(settings, source.id), @@ -2261,9 +2664,10 @@ export default function AddressBookPage({ settings, auth, onAuthChange }: Props) - + + @@ -3167,28 +3571,155 @@ export default function AddressBookPage({ settings, auth, onAuthChange }: Props) setImportOpen(false)} closeDisabled={saving} + className="address-import-dialog" footerClassName="button-row compact-actions" footer={ <> - + {importMode === "vcard" && + } + {importMode === "tabular" && creatingImportProfile && + } + {importMode === "tabular" && !creatingImportProfile && + } + {importMode === "tabular" && !creatingImportProfile && importRun && + } }> -
void submitVcardImport(event)}> -

Paste one or more vCard entries into the selected address book.

- -