Release v0.1.17

.gitea/workflows/module-package-release.yml

@@ -0,0 +1,270 @@
+name: Module Package Release
+
+on:
+  push:
+    tags:
+      - "v*"
+  workflow_dispatch:
+    inputs:
+      release_tag:
+        description: Existing protected version tag to publish
+        required: true
+        type: string
+
+jobs:
+  publish-packages:
+    runs-on: ubuntu-latest
+    env:
+      GITEA_REPOSITORY: ${{ gitea.repository }}
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
+        with:
+          fetch-depth: 0
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065
+        with:
+          python-version: "3.12"
+      - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020
+        with:
+          node-version: "22"
+      - name: Select and validate protected release tag
+        shell: bash
+        env:
+          REQUESTED_TAG: ${{ inputs.release_tag }}
+          TRIGGER_TAG: ${{ gitea.ref_name }}
+        run: |
+          set -euo pipefail
+          tag="${REQUESTED_TAG:-$TRIGGER_TAG}"
+          case "$tag" in
+            v[0-9]*.[0-9]*.[0-9]*) ;;
+            *) echo "Release tag must start with a SemVer-shaped vX.Y.Z value" >&2; exit 1 ;;
+          esac
+          git fetch --force origin "refs/tags/$tag:refs/tags/$tag" refs/heads/main:refs/remotes/origin/main
+          tag_commit="$(git rev-list -n 1 "$tag")"
+          git merge-base --is-ancestor "$tag_commit" refs/remotes/origin/main || {
+            echo "Release tag is not contained in main" >&2
+            exit 1
+          }
+          git checkout --detach "$tag"
+          printf 'RELEASE_TAG=%s\n' "$tag" >> "$GITEA_ENV"
+          printf 'SOURCE_DATE_EPOCH=%s\n' "$(git show -s --format=%ct HEAD)" >> "$GITEA_ENV"
+      - name: Validate package versions
+        run: |
+          python - <<'PY'
+          import json
+          from pathlib import Path
+          import os
+          import re
+          import tomllib
+
+          tag = os.environ["RELEASE_TAG"]
+          expected = tag.removeprefix("v")
+          project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
+          if project.get("version") != expected:
+              raise SystemExit(f"pyproject version {project.get('version')!r} does not match {tag}")
+          if re.fullmatch(r"govoplan-[a-z0-9-]+", str(project.get("name", ""))) is None:
+              raise SystemExit("Python distribution name must use the govoplan-* namespace")
+          webui = Path("webui/package.json")
+          if webui.is_file():
+              package = json.loads(webui.read_text(encoding="utf-8"))
+              if package.get("version") != expected:
+                  raise SystemExit(f"WebUI version {package.get('version')!r} does not match {tag}")
+              if re.fullmatch(r"@govoplan/[a-z0-9-]+-webui", str(package.get("name", ""))) is None:
+                  raise SystemExit("WebUI package name must use the @govoplan/*-webui namespace")
+              release = Path("webui/package.release.json")
+              if release.is_file():
+                  release_package = json.loads(release.read_text(encoding="utf-8"))
+                  if (
+                      release_package.get("name") != package.get("name")
+                      or release_package.get("version") != expected
+                  ):
+                      raise SystemExit("WebUI release package identity does not match package.json and the release tag")
+          PY
+      - name: Build immutable package artifacts
+        shell: bash
+        run: |
+          set -euo pipefail
+          python -m pip install --disable-pip-version-check build==1.5.0 twine==7.0.0
+          rm -rf dist .package-webui
+          python -m build --wheel --outdir dist
+          python -m twine check dist/*.whl
+          if [[ -f webui/package.json ]]; then
+            mkdir .package-webui
+            cp -a webui/. .package-webui/
+            rm -rf .package-webui/node_modules .package-webui/dist
+            if [[ -f .package-webui/package.release.json ]]; then
+              cp .package-webui/package.release.json .package-webui/package.json
+            fi
+          node <<'NODE'
+          const fs = require("node:fs");
+          const path = ".package-webui/package.json";
+          const packageJson = JSON.parse(fs.readFileSync(path, "utf8"));
+          const groups = ["dependencies", "optionalDependencies", "peerDependencies"];
+          for (const group of groups) {
+            for (const [name, specifier] of Object.entries(packageJson[group] || {})) {
+              if (!name.startsWith("@govoplan/")) continue;
+              if (typeof specifier !== "string") {
+                throw new Error(`${group}.${name} must use a string version`);
+              }
+              const packageSlug = name.slice("@govoplan/".length);
+              if (!packageSlug.endsWith("-webui")) {
+                throw new Error(`${group}.${name} is outside the WebUI package namespace`);
+              }
+              const repository = `govoplan-${packageSlug.slice(0, -"-webui".length)}`;
+              const escapedRepository = repository.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
+              const gitTag = specifier.match(
+                new RegExp(
+                  `^git\\+(?:ssh://git@|https://)git\\.add-ideas\\.de/(?:GovOPlaN|add-ideas)/${escapedRepository}\\.git#v([0-9]+\\.[0-9]+\\.[0-9]+)$`,
+                ),
+              );
+              if (gitTag) {
+                packageJson[group][name] = gitTag[1];
+                continue;
+              }
+              if (specifier.startsWith("file:") || specifier.startsWith("git+")) {
+                throw new Error(
+                  `${group}.${name} must resolve to an exact registry version for publication`,
+                );
+              }
+            }
+          }
+          delete packageJson.private;
+          fs.writeFileSync(path, `${JSON.stringify(packageJson, null, 2)}\n`);
+          NODE
+            npm pkg delete private --prefix .package-webui
+            (cd .package-webui && npm pack --ignore-scripts --pack-destination ../dist)
+          fi
+          python - <<'PY'
+          import hashlib
+          import json
+          from pathlib import Path
+          import os
+          import subprocess
+
+          artifacts = []
+          for path in sorted(Path("dist").iterdir()):
+              if path.suffix not in {".whl", ".tgz"}:
+                  continue
+              digest = hashlib.sha256(path.read_bytes()).hexdigest()
+              artifacts.append({"filename": path.name, "sha256": digest, "size": path.stat().st_size})
+          payload = {
+              "schema_version": "1",
+              "repository": os.environ["GITEA_REPOSITORY"],
+              "tag": os.environ["RELEASE_TAG"],
+              "commit": subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(),
+              "artifacts": artifacts,
+          }
+          Path("dist/package-artifacts.json").write_text(
+              json.dumps(payload, indent=2, sort_keys=True) + "\n",
+              encoding="utf-8",
+          )
+          PY
+      - name: Retain package hash evidence
+        uses: actions/upload-artifact@a8a3f3ad30e3422c9c7b888a15615d19a852ae32
+        with:
+          name: module-packages-${{ gitea.ref_name }}
+          path: dist/package-artifacts.json
+      - name: Check immutable registry state
+        shell: bash
+        env:
+          PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
+        run: |
+          set -euo pipefail
+          test -n "$PACKAGE_TOKEN"
+          python - <<'PY'
+          import hashlib
+          import json
+          import os
+          from pathlib import Path
+          import tomllib
+          from urllib.error import HTTPError
+          from urllib.parse import quote
+          from urllib.request import Request, urlopen
+
+          api_root = "https://git.add-ideas.de/api/v1/packages/GovOPlaN"
+          token = os.environ["PACKAGE_TOKEN"]
+
+          def should_publish(kind, name, version, path):
+              package_url = "/".join(
+                  (api_root, kind, quote(name, safe=""), quote(version, safe=""), "files")
+              )
+              request = Request(
+                  package_url,
+                  headers={"Accept": "application/json", "Authorization": f"token {token}"},
+              )
+              try:
+                  with urlopen(request, timeout=30) as response:
+                      files = json.load(response)
+              except HTTPError as exc:
+                  if exc.code == 404:
+                      print(f"{kind} package {name}=={version} is not published yet")
+                      return True
+                  raise
+              if not isinstance(files, list) or len(files) != 1:
+                  raise SystemExit(
+                      f"immutable {kind} package {name}=={version} has an unexpected file set"
+                  )
+              expected_sha256 = hashlib.sha256(path.read_bytes()).hexdigest()
+              if files[0].get("sha256") != expected_sha256:
+                  raise SystemExit(
+                      f"immutable {kind} package {name}=={version} already exists with a different SHA-256"
+                  )
+              print(f"verified existing {kind} package {name}=={version} ({expected_sha256})")
+              return False
+
+          project = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]
+          wheels = tuple(Path("dist").glob("*.whl"))
+          if len(wheels) != 1:
+              raise SystemExit("release build must contain exactly one wheel")
+          publish_pypi = should_publish(
+              "pypi", str(project["name"]), str(project["version"]), wheels[0]
+          )
+
+          tarballs = tuple(Path("dist").glob("*.tgz"))
+          if len(tarballs) > 1:
+              raise SystemExit("release build must contain at most one npm package")
+          publish_npm = False
+          if tarballs:
+              webui = json.loads(
+                  Path(".package-webui/package.json").read_text(encoding="utf-8")
+              )
+              publish_npm = should_publish(
+                  "npm", str(webui["name"]), str(webui["version"]), tarballs[0]
+              )
+
+          with Path(os.environ["GITEA_ENV"]).open("a", encoding="utf-8") as env_file:
+              env_file.write(f"PUBLISH_PYPI={int(publish_pypi)}\n")
+              env_file.write(f"PUBLISH_NPM={int(publish_npm)}\n")
+          PY
+      - name: Publish wheel and WebUI package
+        shell: bash
+        env:
+          PACKAGE_USERNAME: ${{ secrets.GOVOPLAN_PACKAGE_USERNAME }}
+          PACKAGE_TOKEN: ${{ secrets.GOVOPLAN_PACKAGE_TOKEN }}
+        run: |
+          set -euo pipefail
+          test -n "$PACKAGE_USERNAME"
+          test -n "$PACKAGE_TOKEN"
+          if [[ "$PUBLISH_PYPI" == 1 ]]; then
+            TWINE_USERNAME="$PACKAGE_USERNAME" TWINE_PASSWORD="$PACKAGE_TOKEN" \
+              python -m twine upload --non-interactive \
+                --repository-url https://git.add-ideas.de/api/packages/GovOPlaN/pypi \
+                dist/*.whl
+          else
+            echo "Exact wheel is already present; skipping immutable retry."
+          fi
+          shopt -s nullglob
+          webui_packages=(dist/*.tgz)
+          if (( ${#webui_packages[@]} )) && [[ "$PUBLISH_NPM" == 1 ]]; then
+            npmrc="$(mktemp)"
+            trap 'rm -f "$npmrc"' EXIT
+            chmod 600 "$npmrc"
+            printf '%s\n' \
+              '@govoplan:registry=https://git.add-ideas.de/api/packages/GovOPlaN/npm/' \
+              "//git.add-ideas.de/api/packages/GovOPlaN/npm/:_authToken=$PACKAGE_TOKEN" \
+              > "$npmrc"
+            NPM_CONFIG_USERCONFIG="$npmrc" npm publish "./${webui_packages[0]}" \
+              --ignore-scripts --access public \
+              --registry https://git.add-ideas.de/api/packages/GovOPlaN/npm/
+          elif (( ${#webui_packages[@]} )); then
+            echo "Exact WebUI package is already present; skipping immutable retry."
+          fi

README.md

@@ -105,7 +105,8 @@ run profile policy with the `browse` operation, and support Seafile,
 WebDAV/Nextcloud, and SMB when the optional `smb` extra is installed. Seafile
 profiles browse libraries and directories via Seafile's read-only API; profiles
 can still opt into WebDAV browsing by setting `metadata.webdav_endpoint_url` or
-`metadata.browse_protocol` to `webdav`.
+`metadata.browse_protocol` to `webdav`. S3-compatible profiles support bucket
+and prefix browsing when the optional `s3` extra is installed.
 `POST /api/v1/files/connectors/profiles/{profile_id}/import` imports a Seafile
 or WebDAV/Nextcloud/SMB file into managed storage through the same governance,
 conflict handling, source provenance, and connector audit path as direct
@@ -113,17 +114,21 @@ uploads. The Seafile provider uses account-token auth and the native file
 download-link API; Nextcloud and generic WebDAV profiles use authenticated `GET`
 requests against the configured WebDAV endpoint. SMB profiles use
 `smb://server[:port]/share[/path]` endpoints and deployment-owned or encrypted
-stored credentials through `smbprotocol`. Because that SDK cannot accept a
+stored credentials through `smbprotocol`. Files installs a pinned transport for
-preconnected socket and may follow DFS referrals to additional hosts, live SMB
+every initial session, reconnect, alias, and DFS referral target. The deployment
-access fails closed in both public-only and private-network deployments. It will
+private-network policy is re-evaluated immediately before each socket opens.
-remain disabled until every initial connection and referral target can be
-policy-validated and pinned.
 
-S3 connector browse/import remains fail-closed because a user-governed connector
+S3 connector browse/import binds botocore HTTP and HTTPS pools to the same pinned
-endpoint must use the GovOPlaN pinned HTTP trust model. Durable platform storage
+socket policy. Retries, redirects, endpoint discovery, bucket aliases, and new
-has a separate deployment boundary: installer-owned Garage is accepted only at
+connections are therefore revalidated while TLS keeps the configured hostname
-its exact generated endpoint, while an operator-selected external backend
+for SNI and certificate checks. Connector clients do not use outbound proxies or
-requires a clean HTTPS origin and
+ambient AWS credential discovery; configure credentials on the governed profile,
+or explicitly use an anonymous profile for public objects. An incompatible SDK
+upgrade fails closed before a usable client/session is returned.
+
+Durable platform storage has a separate deployment boundary: installer-owned
+Garage is accepted only at its exact generated endpoint, while an
+operator-selected external backend requires a clean HTTPS origin and
 `FILE_STORAGE_S3_ENDPOINT_TRUSTED=true`. That flag is deployment configuration,
 cannot be supplied through a Files connector profile, and does not replace
 operator responsibility for DNS, certificates, egress, bucket policy,
@@ -166,6 +171,12 @@ database and streamed object evidence, while rollback compensates only a newly
 reserved unreferenced key. New object keys are opaque and do not retain the
 uploaded filename. Uncertain or mismatched effects remain visible through Ops.
 
+Operators with `files:file:admin` can run bounded, resumable integrity scans in
+Administration. Scan batches and finding actions carry monotonic revisions;
+stale resume, recheck, or cleanup requests fail before touching object storage.
+Orphan cleanup always requires a dry-run preview followed by separate
+confirmation and records recovery-ledger evidence.
+
 Bulk rename and transfer APIs are owner-scoped: callers must provide the active
 user or group file space with `owner_type` and `owner_id`. The storage layer
 keeps a named legacy file-only helper for historical callers that lack owner

docs/CONNECTOR_BOUNDARY.md

@@ -19,6 +19,7 @@ normal product workflows and can share the same governance model:
 - Nextcloud through WebDAV
 - generic WebDAV
 - SMB through the optional `smb` extra
+- S3-compatible stores through the optional `s3` extra
 
 These providers are surfaced through connector descriptors at
 `GET /api/v1/files/connectors/providers`. Provider descriptors declare whether
@@ -60,11 +61,14 @@ Every provider must:
 - use a transport that pins every connection to a policy-validated DNS/IP answer
   and revalidates redirects; SDK transports without that guarantee fail closed
 
-Live SMB access is disabled in all modes until `smbprotocol` initial connections
+SMB initial connections, reconnects, aliases, and DFS referral targets are
-and DFS referral targets can be pinned and policy-validated. An explicit IP is
+created through a Files-owned pinned `smbprotocol` transport. S3 HTTP/HTTPS pools
-not sufficient because the server may still issue a referral to another peer.
+use the equivalent botocore adapter for every connection selected by retries,
-Live S3 access is likewise disabled until `boto3`/botocore can be bound to the
+redirects, endpoint discovery, and virtual-host addressing. Both adapters apply
-pinned transport, including SDK-managed redirects and endpoint discovery.
+the deployment-wide private-network policy at socket creation. S3 keeps the
+configured hostname for TLS SNI and certificate verification, but does not use
+outbound proxies or ambient AWS credential discovery. An incompatible optional
+SDK release fails closed before a usable session or client is returned.
 
 ## Non-Goals For Files
 

docs/FILES_HANDBOOK.md

@@ -405,15 +405,15 @@ in a development/test runtime.
 | Seafile | Read-only native API browse/download-link import and manual sync implemented using the pinned HTTP transport; WebDAV opt-in supported |
 | Nextcloud | Read-only WebDAV browse/import/manual sync implemented using the pinned HTTP transport |
 | Generic WebDAV | Read-only browse/import/manual sync implemented using the pinned HTTP transport |
-| SMB | Browse/import code and descriptor implemented, but every live connection fails closed until initial connections and DFS referrals can be policy-validated and pinned |
+| SMB | Read-only browse/import/manual sync implemented through a pinned smbprotocol transport for initial peers, reconnects, aliases, and DFS referral targets |
-| S3 connector | Browse/import code and descriptor implemented, but every live SDK connection fails closed until botocore connections, redirects, and endpoint discovery can be validated and pinned |
+| S3 connector | Read-only bucket/prefix browse, import, and manual sync implemented through pinned botocore pools covering retries, redirects, endpoint discovery, and provider aliases |
 | SharePoint and OneDrive | Provider keys/descriptors reserved; live Microsoft Graph browse/import is planned |
 | NFS and local connector | Described as optional future providers; the local managed-storage backend is a different feature |
 
 Provider descriptors are available from
 `GET /api/v1/files/connectors/providers`. Use their `implemented`, `installed`,
-and support fields for display, but expect a fail-closed transport error where
+and support fields for display. An incompatible optional SDK release fails closed
-the table above says live access is disabled.
+before it can return a usable client or session.
 
 ## Operator runbook
 
@@ -478,17 +478,29 @@ The built-in HTTP transport:
 - bounds structured responses to 16 MiB and file transfers to 512 MiB by
   default.
 
+The S3 SDK adapter applies the same socket rule to every botocore pool selected
+for a retry, redirect, discovered endpoint, or virtual-host bucket alias. The
+original authority remains in the request and TLS SNI/certificate check. S3
+connector clients use no outbound proxy and never discover ambient AWS
+credentials: configure both access and secret keys on the governed profile, or
+use an anonymous profile for a public source.
+
+The SMB adapter owns a separate connection cache and replaces smbprotocol's TCP
+factory process-wide with the stricter pinned socket. Initial peers, reconnects,
+server aliases, domain-controller connections, and DFS referral targets therefore
+pass the same policy at connection time. Signing is required by default; enable
+SMB encryption on the profile where the server supports it.
+
 Override the connector response limits with
 `GOVOPLAN_CONNECTOR_MAX_STRUCTURED_RESPONSE_BYTES` and
 `GOVOPLAN_CONNECTOR_MAX_FILE_TRANSFER_BYTES`. The smaller applicable limit wins
 when an import is also subject to `FILE_UPLOAD_MAX_BYTES`.
 
 Never work around a connector pinning failure by adding a raw IP, disabling TLS,
-or enabling private networks. SMB may redirect through DFS, and user-configured
+or enabling private networks. A failure means the peer policy rejected an actual
-S3 connectors may perform their own redirects or endpoint discovery; those
+connection destination or the installed SDK no longer exposes the verified
-connector transports remain disabled until every connection peer can be
+transport seam. The separately configured platform S3 backend is trusted only by
-governed. The separately configured platform S3 backend is trusted only by the
+the deployment owner and is not selectable by a user or connector profile.
-deployment owner and is not selectable by a user or connector profile.
 
 ### Backup and restore
 
@@ -509,10 +521,13 @@ snapshots so database references and objects represent the same recovery point.
 The integrity API verifies a restored set, but it does not replace a coordinated
 backup.
 
-Create a scan with `POST /api/v1/files/integrity/scans`, then call
+Operators normally use **Administration > File integrity**. The equivalent API
-`POST /api/v1/files/integrity/scans/{scan_id}/run` until it reports
+creates a scan with `POST /api/v1/files/integrity/scans`, then calls
-`completed`. Each call advances at most the persisted batch size, so a stopped
+`POST /api/v1/files/integrity/scans/{scan_id}/run` with the scan's current
-operator or worker can resume from the committed blob/object cursors.
+`expected_revision` until it reports `completed`. Each call advances at most
+the persisted batch size, so a stopped operator or worker can resume from the
+committed blob/object cursors. Concurrent or stale actions receive `409` before
+the storage backend is invoked; reload the scan and inspect the newer state.
 
 Findings distinguish:
 
@@ -524,10 +539,14 @@ Findings distinguish:
 
 Missing or corrupt blobs fail closed for ordinary downloads and Campaign
 attachment materialization. After restoring the expected bytes, use the finding
-`recheck` action first in dry-run mode and then apply it. Orphan cleanup is also
+`recheck` action with its current `expected_revision`. Orphan cleanup starts
-dry-run by default, rechecks that no database reference exists, remains scoped
+with a dry-run preview and requires separate destructive confirmation. The
-to the scanned tenant prefix, and is idempotent. Both applied and dry-run
+confirmation reuses the finding revision from that preview, rechecks that no
-actions emit audit evidence.
+database reference exists, remains scoped to the scanned tenant prefix, and is
+idempotent. Both applied and dry-run actions emit audit evidence. A shared
+reference blocks deletion. Files currently has no legal-hold or hard-purge
+model, so retention-controlled objects must not be treated as cleanup
+candidates until those controls are implemented.
 
 ### Recovery ledger for object effects
 
@@ -613,7 +632,7 @@ Use these symptoms as routing hints:
 | `Stored object does not exist` | Database/blob restore mismatch, wrong root, or missing shared storage |
 | `Stored secret cannot be decrypted` | Wrong or rotated `MASTER_KEY_B64` |
 | Private/non-public endpoint blocked | Deployment-wide egress policy is public-only or DNS returned a forbidden answer |
-| SDK cannot pin redirects/referrals | Expected fail-closed S3/SMB boundary, not a transient connector outage |
+| SDK peer-pinning seam is unavailable | Optional S3/SMB SDK is incompatible; keep access fail-closed and validate the supported dependency range before upgrade |
 | Connector response exceeds limit | Remote payload exceeds connector or upload limit |
 | Profile is not visible | Scope, disabled state, campaign access, or policy mismatch |
 | Group removal is vetoed | The group still owns a file/folder/connector space or is a share target |
@@ -817,9 +836,11 @@ bypass the same rule.
 ### Pinned connector transport
 
 Given public-only mode and DNS returns any private address, the connection must
-be rejected before a socket opens. Given private mode, the HTTP connection must
+be rejected before a socket opens. Given private mode, every HTTP, S3, and SMB
-still use the validated address and reject redirects. SMB and S3 must fail
+connection must still use an address from the answer validated for that exact
-before their SDK clients connect until all SDK-managed peers can be pinned.
+attempt. Botocore retries, redirects, endpoint discovery, and aliases, plus SMB
+reconnects and DFS referrals, must pass through the pinned factories. A changed
+or unsupported SDK seam must fail before a usable client/session is returned.
 
 ### Imported evidence and sync
 
@@ -860,7 +881,7 @@ returning different content or credentials.
 | Deletion/retention | Soft-delete assets/folders/spaces; immediate audited connector-secret scrubbing | File restore API, hard purge, retention policy, legal hold, and blob GC ([#38](https://git.add-ideas.de/GovOPlaN/govoplan-files/issues/38)) |
 | Connector governance | Scoped profiles/credentials/policies, effective source explanation, separate credentials, linked user/group spaces | Provider-owned external secret lifecycle; API `secret_ref` remains rejected |
 | HTTP connectors | Pinned, bounded, no-redirect Seafile and WebDAV/Nextcloud browse/import/manual sync | Background/folder sync, remote mutation, long-running transfer workers |
-| SMB and S3 connectors | Provider descriptors and browse/import logic | Live access only after SDK connections plus DFS referrals/redirects are validated and pinned ([SMB #35](https://git.add-ideas.de/GovOPlaN/govoplan-files/issues/35), [S3 #34](https://git.add-ideas.de/GovOPlaN/govoplan-files/issues/34)) |
+| SMB and S3 connectors | Provider descriptors, browse/import/manual sync, pinned SDK transports, and redirect/retry/referral transport-contract tests | Live topology smoke evidence, provider-specific OAuth, remote writes, and background indexing remain separate deployment or connector-module concerns |
 | Other providers | Reserved SharePoint/OneDrive keys and NFS/local descriptors | Graph/OAuth/provider paging, NFS deployment integration, DMS connectors |
 | Connector spaces | User/group link, browse, manual selected-file sync, edit/disable/delete | Background sync, remote writes/deletes, full conflict-reporting jobs |
 | Profile capabilities | Stored and displayed | Enforce capability flags as an independent operation gate |
@@ -883,8 +904,8 @@ Before releasing Files:
 6. Exercise upload, ZIP bounds, conflict handling, download, and soft deletion.
 7. Exercise connector policy explanation and one pinned HTTP provider where
    configured; verify the deployment-managed or trusted external S3 backend if
-   selected, and verify user-configured S3 and SMB connector peers still fail
+   selected, and verify one configured S3 and SMB connector while recording the
-   closed.
+   actual target topology and private-network policy.
 8. Verify credential deletion scrubs dependents and produces audit evidence.
 9. Verify a campaign attachment snapshot still identifies its exact version and
    checksum after the current file changes.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/files-webui",
-  "version": "0.1.9",
+  "version": "0.1.17",
   "private": true,
   "type": "module",
   "main": "webui/src/index.ts",
@@ -19,14 +19,14 @@
     "LICENSE"
   ],
   "peerDependencies": {
-    "@govoplan/core-webui": "^0.1.9",
+    "@vitejs/plugin-react": "^5.2.0",
-    "lucide-react": "^1.23.0",
+    "vite": "^7.3.6",
-    "react": "^19.0.0",
-    "react-dom": "^19.0.0",
-    "react-router-dom": "^7.1.1",
-    "@vitejs/plugin-react": "^4.3.4",
     "typescript": "^5.7.2",
-    "vite": "^6.0.6"
+    "react": ">=19.2.7 <20",
+    "react-dom": ">=19.2.7 <20",
+    "react-router": ">=8.3.0 <9",
+    "lucide-react": "^1.23.0",
+    "@govoplan/core-webui": "^0.1.17"
   },
   "peerDependenciesMeta": {
     "@govoplan/core-webui": {

pyproject.toml

@@ -4,14 +4,14 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "govoplan-files"
-version = "0.1.9"
+version = "0.1.17"
 description = "GovOPlaN files module with backend and WebUI integration."
 readme = "README.md"
 requires-python = ">=3.12"
 license = { file = "LICENSE" }
 authors = [{ name = "GovOPlaN" }]
 dependencies = [
-  "govoplan-core>=0.1.14",
+  "govoplan-core>=0.1.17",
   "defusedxml>=0.7,<1",
   "pyzipper>=0.3.6,<1",
   "python-multipart>=0.0.31,<1",

src/govoplan_files/backend/db/models.py

@@ -54,6 +54,7 @@ class FileIntegrityScan(Base, TimestampMixin):
     storage_backend: Mapped[str] = mapped_column(String(50), nullable=False)
     storage_prefix: Mapped[str] = mapped_column(String(1000), nullable=False)
     status: Mapped[str] = mapped_column(String(30), default="pending", nullable=False, index=True)
+    revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
     phase: Mapped[str] = mapped_column(String(30), default="blobs", nullable=False)
     verify_checksums: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
     batch_size: Mapped[int] = mapped_column(Integer, default=100, nullable=False)
@@ -81,6 +82,7 @@ class FileIntegrityFinding(Base, TimestampMixin):
     tenant_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
     kind: Mapped[str] = mapped_column(String(40), nullable=False, index=True)
     state: Mapped[str] = mapped_column(String(30), default="open", nullable=False, index=True)
+    revision: Mapped[int] = mapped_column(Integer, default=1, nullable=False)
     blob_id: Mapped[str | None] = mapped_column(ForeignKey("file_blobs.id", ondelete="SET NULL"), nullable=True, index=True)
     storage_key: Mapped[str] = mapped_column(String(1000), nullable=False)
     expected_size_bytes: Mapped[int | None] = mapped_column(Integer, nullable=True)

src/govoplan_files/backend/manifest.py

@@ -35,6 +35,7 @@ from govoplan_core.core.provider_governance import (
     ProviderObjectDeclaration,
     declared_module_architecture,
 )
+from govoplan_core.core.search import SearchSourceProviderRegistration
 from govoplan_core.core.views import ViewSurface
 from govoplan_core.db.base import Base
 from govoplan_files.backend.change_tracking import register_files_change_tracking
@@ -44,6 +45,7 @@ from govoplan_files.backend.provider_state import (
     REMOTE_STORAGE_PROVIDER_ID,
     remote_storage_provider_states,
 )
+from govoplan_files.backend.search_source import create_files_search_source
 
 register_files_change_tracking()
 
@@ -277,9 +279,9 @@ REMOTE_STORAGE_PROVIDER = ExternalProviderDeclaration(
 manifest = ModuleManifest(
     id="files",
     name="Files",
-    version="0.1.9",
+    version="0.1.17",
     required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
-    optional_dependencies=("campaigns", "encryption"),
+    optional_dependencies=("campaigns", "encryption", "search"),
     provides_interfaces=(
         ModuleInterfaceProvider(name="files.access", version="0.1.6"),
         ModuleInterfaceProvider(name="files.campaign_attachments", version="0.1.6"),
@@ -298,12 +300,24 @@ manifest = ModuleManifest(
             version_max_exclusive="2.0.0",
             optional=True,
         ),
+        ModuleInterfaceRequirement(
+            name="search.source",
+            version_min="1.0.0",
+            version_max_exclusive="2.0.0",
+            optional=True,
+        ),
     ),
     permissions=PERMISSIONS,
     route_factory=_files_router,
     role_templates=ROLE_TEMPLATES,
     tenant_summary_providers=(_tenant_summary,),
     tenant_summary_batch_providers=(_tenant_summary_batch,),
+    search_sources=(
+        SearchSourceProviderRegistration(
+            id="files.objects",
+            factory=create_files_search_source,
+        ),
+    ),
     delete_veto_providers={"group": (_veto_group_delete,)},
     nav_items=(NavItem(path="/files", label="Files", icon="folder", required_any=("files:file:read",), order=40),),
     frontend=FrontendModule(
@@ -321,6 +335,7 @@ manifest = ModuleManifest(
         view_surfaces=(
             ViewSurface(id="files.admin.system-connectors", module_id="files", kind="section", label="System file connections", order=75),
             ViewSurface(id="files.admin.tenant-connectors", module_id="files", kind="section", label="Tenant file connections", order=65),
+            ViewSurface(id="files.admin.tenant-integrity", module_id="files", kind="section", label="File integrity", order=66),
             ViewSurface(id="files.admin.group-connectors", module_id="files", kind="section", label="Group file connections", order=65),
             ViewSurface(id="files.admin.user-connectors", module_id="files", kind="section", label="User file connections", order=65),
             ViewSurface(id="files.settings.connectors", module_id="files", kind="section", label="Personal file connections", order=20),
@@ -328,6 +343,40 @@ manifest = ModuleManifest(
         ),
     ),
     documentation=(
+        DocumentationTopic(
+            id="files.search.managed-content",
+            title="Search managed files and folders",
+            summary="Expose file names, logical paths, and descriptions to permission-aware platform Search.",
+            body=(
+                "When Search is installed, Files contributes managed files and folders to its derived index. "
+                "Every result is tenant-bounded and rechecks current ownership, group membership, direct shares, "
+                "expiry, revocation, deletion, and Files permissions before it is returned. Committed file and "
+                "share changes are delivered through the platform event outbox; an administrator can rebuild the "
+                "derived index without changing authoritative Files data."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("file_user", "file_manager", "administrator"),
+            related_modules=("search",),
+            order=41,
+            conditions=(
+                DocumentationCondition(
+                    required_modules=("files",),
+                    any_scopes=("files:file:read", "files:file:admin"),
+                ),
+            ),
+            links=(
+                DocumentationLink(label="Files", href="/files", kind="runtime"),
+                DocumentationLink(label="Search", href="/search", kind="runtime"),
+                DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository"),
+            ),
+            metadata={
+                "kind": "reference",
+                "route": "/files",
+                "screen": "Files search contribution",
+                "help_contexts": ["files.list"],
+            },
+        ),
         DocumentationTopic(
             id="files.workflow.organize-managed-files",
             title="Organize managed files and folders",
@@ -586,10 +635,10 @@ manifest = ModuleManifest(
         DocumentationTopic(
             id="files.reference.integrity-recovery-and-fail-closed-transports",
             title="Operate Files integrity, recovery, and connector transport safety",
-            summary="Back up database evidence, blob ciphertext, and Encryption custody as one recovery unit, and keep unsupported SDK transports fail-closed.",
+            summary="Back up database evidence, blob ciphertext, and Encryption custody as one recovery unit, and pin every SDK-managed connector peer.",
             body=(
-                "Local durable storage is the operational baseline. Recover Files from a coordinated database/blob snapshot with the matching Encryption tables and original deployment master key, then run the bounded resumable integrity scan and verify representative protected and unprotected access paths. Protected scans verify stored ciphertext before decryption and then verify plaintext semantic evidence. Managed blob creation/repair and applied orphan cleanup commit lease-fenced Core recovery intent before object effects; success, compensation, and forward completion require independent database and object checks, while mismatch is quarantined and unresolved work remains visible in Ops. Missing or mismatched blobs are quarantined; orphan objects are reported before dry-run-first, explicitly authorized cleanup. "
+                "Local durable storage is the operational baseline. Recover Files from a coordinated database/blob snapshot with the matching Encryption tables and original deployment master key, then run the bounded resumable integrity scan from Administration and verify representative protected and unprotected access paths. Each scan batch and finding action requires the revision shown to the operator, so a stale screen cannot recheck or delete after concurrent reconciliation. Protected scans verify stored ciphertext before decryption and then verify plaintext semantic evidence. Managed blob creation/repair and applied orphan cleanup commit lease-fenced Core recovery intent before object effects; success, compensation, and forward completion require independent database and object checks, while mismatch is quarantined and unresolved work remains visible in Ops. Missing or mismatched blobs are quarantined; orphan objects are reported before dry-run-first, explicitly authorized cleanup. "
-                "Arbitrary external S3 managed storage/connectors and SMB connectors fail closed until botocore redirects/endpoint discovery and SMB initial connections/DFS referrals support connection-time DNS/IP pinning. Installer-owned Garage storage is supported only at the exact deployment service endpoint with its explicit trust marker. Destructive module retirement drops database tables but does not remove backend blob objects."
+                "S3 connector pools pin every retry, redirect, discovered endpoint, and provider alias while retaining the configured TLS authority; outbound proxies and ambient credential discovery are disabled. SMB initial connections, reconnects, aliases, and DFS referrals use a Files-owned pinned transport and cache. Both apply the deployment private-network policy immediately before each socket opens and fail closed if an SDK no longer exposes the verified transport seam. Installer-owned Garage storage is supported only at the exact deployment service endpoint with its explicit trust marker. Destructive module retirement drops database tables but does not remove backend blob objects."
             ),
             layer="configured",
             documentation_types=("admin",),
@@ -608,6 +657,7 @@ manifest = ModuleManifest(
             ),
             links=(
                 DocumentationLink(label="System file connections", href="/admin?section=system-file-connectors", kind="runtime"),
+                DocumentationLink(label="File integrity operations", href="/admin?section=tenant-file-integrity", kind="runtime"),
                 DocumentationLink(label="Connector provider status", href="/api/v1/files/connectors/providers", kind="api"),
                 DocumentationLink(label="Create an integrity scan", href="/api/v1/files/integrity/scans", kind="api"),
                 DocumentationLink(label="Files handbook", href="govoplan-files/docs/FILES_HANDBOOK.md", kind="repository"),
@@ -636,7 +686,7 @@ manifest = ModuleManifest(
                 "screen": "System file connections and deployment operations",
                 "section": "Storage integrity, backup/recovery, and fail-closed transports",
                 "recovery_unit": ["Files database rows", "Encryption envelope and wrapped-key rows", "managed blob namespace", "MASTER_KEY_B64", "deployment-owned connector configuration"],
-                "verification": "After restore, complete a checksum-enabled integrity scan, resolve every missing/corrupt finding, approve or retain every reported orphan, inspect Files recovery operations in Ops, verify authorized and denied access, and test one permitted pinned HTTP connector.",
+                "verification": "After restore, complete a checksum-enabled integrity scan, resolve every missing/corrupt finding, approve or retain every reported orphan, inspect Files recovery operations in Ops, verify authorized and denied access, and test configured pinned HTTP, S3, and SMB connectors against their recorded target topology.",
                 "related_topic_ids": [
                     "files.governed-connectors-and-provenance",
                     "files.reference.snapshot-provenance-and-capabilities",
@@ -741,7 +791,7 @@ manifest = ModuleManifest(
                     "Confirm version alignment, apply migrations on clean and upgrade databases, and run the repository plus meta-repository security and static-analysis gates.",
                     "Exercise allowed and denied personal, group, and share access with representative accounts.",
                     "Exercise bounded archive preview and confirmation, every required conflict strategy, organization, download, and soft deletion.",
-                    "Where connectors are configured, verify policy explanation and one pinned HTTP provider; verify installer-owned Garage when selected, and verify arbitrary external S3 and SMB access still fails closed.",
+                    "Where connectors are configured, verify policy explanation and one pinned HTTP provider; verify installer-owned Garage when selected, then exercise configured S3 retries/aliases and SMB reconnect/referral targets under the deployment private-network policy, confirming an incompatible SDK transport seam fails closed.",
                     "Restore a coordinated database/blob/key backup and compare representative downloaded bytes with their recorded SHA-256 checksums.",
                     "Record the tested versions, results, known limitations, evidence locations, residual risks, owner, and approval decision.",
                 ],

src/govoplan_files/backend/migrations/versions/f1a2b3c4d5e7_file_integrity_action_revisions.py

@@ -0,0 +1,34 @@
+"""add stale-action revisions to Files integrity operations
+
+Revision ID: f1a2b3c4d5e7
+Revises: d0e1f2a3b4c6
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = "f1a2b3c4d5e7"
+down_revision = "d0e1f2a3b4c6"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("file_integrity_scans") as batch_op:
+        batch_op.add_column(
+            sa.Column("revision", sa.Integer(), nullable=False, server_default="1")
+        )
+    with op.batch_alter_table("file_integrity_findings") as batch_op:
+        batch_op.add_column(
+            sa.Column("revision", sa.Integer(), nullable=False, server_default="1")
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("file_integrity_findings") as batch_op:
+        batch_op.drop_column("revision")
+    with op.batch_alter_table("file_integrity_scans") as batch_op:
+        batch_op.drop_column("revision")

src/govoplan_files/backend/routes/integrity.py

@@ -18,6 +18,7 @@ from govoplan_files.backend.schemas import (
     FileIntegrityFindingResponse,
     FileIntegrityFindingsResponse,
     FileIntegrityScanCreateRequest,
+    FileIntegrityScanRunRequest,
     FileIntegrityScanResponse,
     FileIntegrityScansResponse,
 )
@@ -93,10 +94,17 @@ def create_scan(
 @router.post("/scans/{scan_id}/run", response_model=FileIntegrityScanResponse)
 def run_scan_batch(
     scan_id: str,
+    payload: FileIntegrityScanRunRequest,
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(require_scope("files:file:admin")),
 ) -> FileIntegrityScanResponse:
-    scan = _scan_for_tenant(session, scan_id, principal.tenant_id)
+    scan = _scan_for_tenant(
+        session,
+        scan_id,
+        principal.tenant_id,
+        for_update=True,
+    )
+    _assert_expected_revision(scan.revision, payload.expected_revision)
     previous_status = scan.status
     try:
         run_integrity_scan_batch(session, scan)
@@ -117,7 +125,12 @@ def run_scan_batch(
         return _scan_response(scan)
     except (FileStorageError, StorageBackendError) as exc:
         session.rollback()
-        scan = _scan_for_tenant(session, scan_id, principal.tenant_id)
+        scan = _scan_for_tenant(
+            session,
+            scan_id,
+            principal.tenant_id,
+            for_update=True,
+        )
         mark_integrity_scan_failed(scan, error=exc)
         audit_from_principal(
             session,
@@ -172,7 +185,13 @@ def recheck_finding(
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(require_scope("files:file:admin")),
 ) -> FileIntegrityActionResponse:
-    finding = _finding_for_tenant(session, finding_id, principal.tenant_id)
+    finding = _finding_for_tenant(
+        session,
+        finding_id,
+        principal.tenant_id,
+        for_update=True,
+    )
+    _assert_expected_revision(finding.revision, payload.expected_revision)
     try:
         result = recheck_integrity_finding(
             session,
@@ -198,7 +217,13 @@ def cleanup_finding(
     session: Session = Depends(get_session),
     principal: ApiPrincipal = Depends(require_scope("files:file:admin")),
 ) -> FileIntegrityActionResponse:
-    finding = _finding_for_tenant(session, finding_id, principal.tenant_id)
+    finding = _finding_for_tenant(
+        session,
+        finding_id,
+        principal.tenant_id,
+        for_update=True,
+    )
+    _assert_expected_revision(finding.revision, payload.expected_revision)
     try:
         result = cleanup_orphan_finding(
             session,
@@ -218,8 +243,13 @@ def _scan_for_tenant(
     session: Session,
     scan_id: str,
     tenant_id: str,
+    *,
+    for_update: bool = False,
 ) -> FileIntegrityScan:
-    scan = session.get(FileIntegrityScan, scan_id)
+    query = session.query(FileIntegrityScan).filter(FileIntegrityScan.id == scan_id)
+    if for_update:
+        query = query.populate_existing().with_for_update()
+    scan = query.one_or_none()
     if scan is None or scan.tenant_id != tenant_id:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -232,8 +262,15 @@ def _finding_for_tenant(
     session: Session,
     finding_id: str,
     tenant_id: str,
+    *,
+    for_update: bool = False,
 ) -> FileIntegrityFinding:
-    finding = session.get(FileIntegrityFinding, finding_id)
+    query = session.query(FileIntegrityFinding).filter(
+        FileIntegrityFinding.id == finding_id
+    )
+    if for_update:
+        query = query.populate_existing().with_for_update()
+    finding = query.one_or_none()
     if finding is None or finding.tenant_id != tenant_id:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -242,6 +279,17 @@ def _finding_for_tenant(
     return finding
 
 
+def _assert_expected_revision(current: int, expected: int) -> None:
+    if current != expected:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail=(
+                "The integrity record changed after it was loaded; reload before "
+                "performing this action."
+            ),
+        )
+
+
 def _audit_integrity_action(session, principal, result) -> None:
     audit_from_principal(
         session,
@@ -271,6 +319,7 @@ def _scan_response(scan: FileIntegrityScan) -> FileIntegrityScanResponse:
         storage_backend=scan.storage_backend,
         storage_prefix=scan.storage_prefix,
         status=scan.status,
+        revision=scan.revision,
         phase=scan.phase,
         verify_checksums=scan.verify_checksums,
         batch_size=scan.batch_size,
@@ -299,6 +348,7 @@ def _finding_response(
         tenant_id=finding.tenant_id,
         kind=finding.kind,
         state=finding.state,
+        revision=finding.revision,
         blob_id=finding.blob_id,
         storage_key=finding.storage_key,
         expected_size_bytes=finding.expected_size_bytes,

src/govoplan_files/backend/schemas.py

@@ -101,6 +101,7 @@ class FileIntegrityScanResponse(BaseModel):
     storage_backend: str
     storage_prefix: str
     status: str
+    revision: int
     phase: str
     verify_checksums: bool
     batch_size: int
@@ -127,6 +128,7 @@ class FileIntegrityFindingResponse(BaseModel):
     tenant_id: str
     kind: str
     state: str
+    revision: int
     blob_id: str | None = None
     storage_key: str
     expected_size_bytes: int | None = None
@@ -145,6 +147,11 @@ class FileIntegrityFindingsResponse(BaseModel):
 
 class FileIntegrityActionRequest(BaseModel):
     dry_run: bool = True
+    expected_revision: int = Field(ge=1)
+
+
+class FileIntegrityScanRunRequest(BaseModel):
+    expected_revision: int = Field(ge=1)
 
 
 class FileIntegrityActionResponse(BaseModel):

src/govoplan_files/backend/search_source.py

@@ -0,0 +1,329 @@
+from __future__ import annotations
+
+from collections.abc import Mapping, Sequence
+from urllib.parse import quote
+
+from sqlalchemy import func, or_, select
+from sqlalchemy.orm import Session
+
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.events import PlatformEvent
+from govoplan_core.core.modules import ModuleContext
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillPage,
+    SearchBackfillRequest,
+    SearchDocument,
+    SearchIndexChange,
+    SearchResourceReference,
+    SearchResourceType,
+)
+from govoplan_files.backend.db.models import FileAsset, FileFolder, FileShare
+from govoplan_files.backend.storage.share_state import effective_file_share_clause
+
+
+PROVIDER_ID = "files.objects"
+RESOURCE_MODELS = {
+    "file": FileAsset,
+    "folder": FileFolder,
+}
+READ_SCOPE = "files:file:read"
+ADMIN_SCOPE = "files:file:admin"
+
+
+class FilesSearchSource:
+    def resource_types(self) -> Sequence[SearchResourceType]:
+        return (
+            SearchResourceType(
+                provider_id=PROVIDER_ID,
+                module_id="files",
+                resource_type="file",
+                label="Files",
+                requires_authorization_recheck=True,
+            ),
+            SearchResourceType(
+                provider_id=PROVIDER_ID,
+                module_id="files",
+                resource_type="folder",
+                label="File folders",
+                requires_authorization_recheck=True,
+            ),
+        )
+
+    def backfill(
+        self,
+        session: object,
+        *,
+        request: SearchBackfillRequest,
+    ) -> SearchBackfillPage:
+        db = _session(session)
+        model = _model(request.provider_id, request.resource_type)
+        statement = select(model).where(
+            model.tenant_id == request.tenant_id,
+            model.deleted_at.is_(None),
+        )
+        if request.cursor:
+            statement = statement.where(model.id > request.cursor)
+        rows = list(
+            db.scalars(
+                statement.order_by(model.id).limit(request.limit + 1)
+            )
+        )
+        has_more = len(rows) > request.limit
+        selected = rows[: request.limit]
+        shares = (
+            _shares_by_asset(db, selected)
+            if request.resource_type == "file"
+            else {}
+        )
+        high_watermark = db.scalar(
+            select(func.max(model.updated_at)).where(
+                model.tenant_id == request.tenant_id,
+                model.deleted_at.is_(None),
+            )
+        )
+        return SearchBackfillPage(
+            documents=tuple(
+                _document(
+                    row,
+                    resource_type=request.resource_type,
+                    shares=shares.get(row.id, ()),
+                )
+                for row in selected
+            ),
+            next_cursor=selected[-1].id if has_more and selected else None,
+            complete=not has_more,
+            high_watermark=(
+                high_watermark.isoformat()
+                if high_watermark is not None
+                else None
+            ),
+        )
+
+    def authorize(
+        self,
+        session: object,
+        principal: object,
+        *,
+        requests: Sequence[SearchAuthorizationRequest],
+    ) -> Mapping[str, bool]:
+        decisions = {item.reference.key: False for item in requests}
+        if not isinstance(principal, ApiPrincipal) or not (
+            principal.has(READ_SCOPE) or principal.has(ADMIN_SCOPE)
+        ):
+            return decisions
+        db = _session(session)
+        for request in requests:
+            reference = request.reference
+            if (
+                reference.tenant_id != principal.tenant_id
+                or reference.module_id != "files"
+                or reference.resource_type not in RESOURCE_MODELS
+            ):
+                continue
+            decisions[reference.key] = _can_read(
+                db,
+                principal,
+                reference=reference,
+            )
+        return decisions
+
+    def index_changes_for_event(
+        self,
+        session: object,
+        *,
+        event: PlatformEvent,
+        delivery_key: str,
+    ) -> Sequence[SearchIndexChange]:
+        if (
+            event.module_id != "files"
+            or event.tenant is None
+            or event.resource is None
+            or event.resource.id is None
+            or event.resource.type not in RESOURCE_MODELS
+        ):
+            return ()
+        db = _session(session)
+        reference = SearchResourceReference(
+            tenant_id=event.tenant.id,
+            module_id="files",
+            resource_type=event.resource.type,
+            resource_id=event.resource.id,
+        )
+        model = RESOURCE_MODELS[event.resource.type]
+        row = db.get(model, event.resource.id)
+        deleted = row is None or row.tenant_id != event.tenant.id or row.deleted_at is not None
+        cursor = event.event_id
+        document = None
+        if not deleted:
+            shares = (
+                tuple(_active_shares(db, row.id))
+                if event.resource.type == "file"
+                else ()
+            )
+            document = _document(
+                row,
+                resource_type=event.resource.type,
+                shares=shares,
+                change_cursor=cursor,
+            )
+        return (
+            SearchIndexChange(
+                change_id=f"{delivery_key}:{PROVIDER_ID}:{event.resource.type}",
+                provider_id=PROVIDER_ID,
+                kind="delete" if deleted else "upsert",
+                reference=reference,
+                source_revision=(
+                    document.source_revision if document is not None else cursor
+                ),
+                cursor=cursor,
+                document=document,
+                occurred_at=event.occurred_at,
+            ),
+        )
+
+
+def create_files_search_source(_context: ModuleContext) -> FilesSearchSource:
+    return FilesSearchSource()
+
+
+def _model(provider_id: str, resource_type: str):
+    if provider_id != PROVIDER_ID or resource_type not in RESOURCE_MODELS:
+        raise ValueError("Unsupported Files search source.")
+    return RESOURCE_MODELS[resource_type]
+
+
+def _document(
+    row: FileAsset | FileFolder,
+    *,
+    resource_type: str,
+    shares: Sequence[FileShare] = (),
+    change_cursor: str | None = None,
+) -> SearchDocument:
+    is_file = isinstance(row, FileAsset)
+    title = row.filename if is_file else (row.path.rsplit("/", 1)[-1] or row.path)
+    path = row.display_path if is_file else row.path
+    owner_id = row.owner_user_id if row.owner_type == "user" else row.owner_group_id
+    tokens = [f"scope:{READ_SCOPE}", f"scope:{ADMIN_SCOPE}"]
+    if owner_id:
+        tokens.append(
+            f"membership:{owner_id}"
+            if row.owner_type == "user"
+            else f"group:{owner_id}"
+        )
+    for share in shares:
+        prefix = "membership" if share.target_type == "user" else share.target_type
+        if prefix in {"membership", "group", "tenant"}:
+            tokens.append(f"{prefix}:{share.target_id}")
+    updated_at = row.updated_at or row.created_at
+    revision = (
+        f"{row.current_version_id or 'none'}:{updated_at.isoformat()}"
+        if is_file
+        else updated_at.isoformat()
+    )
+    return SearchDocument(
+        tenant_id=row.tenant_id,
+        module_id="files",
+        provider_id=PROVIDER_ID,
+        resource_type=resource_type,
+        resource_id=row.id,
+        title=title,
+        url=f"/files?{resource_type}Id={quote(row.id, safe='')}",
+        summary=((row.description or "") if is_file else path)[:4000] or None,
+        body=" ".join(
+            value for value in (path, row.description if is_file else None) if value
+        )[:200_000],
+        keywords=(path[:200], row.owner_type[:200]),
+        visibility="restricted",
+        acl_tokens=tuple(dict.fromkeys(tokens)),
+        metadata={
+            "path": path,
+            "owner_type": row.owner_type,
+            "current_version_id": row.current_version_id if is_file else None,
+        },
+        source_revision=revision,
+        change_cursor=change_cursor,
+        source_updated_at=updated_at,
+        requires_authorization_recheck=True,
+    )
+
+
+def _can_read(
+    session: Session,
+    principal: ApiPrincipal,
+    *,
+    reference: SearchResourceReference,
+) -> bool:
+    model = RESOURCE_MODELS[reference.resource_type]
+    row = session.get(model, reference.resource_id)
+    if row is None or row.tenant_id != principal.tenant_id or row.deleted_at is not None:
+        return False
+    if principal.has(ADMIN_SCOPE):
+        return True
+    user_id = str(getattr(principal.user, "id", "") or principal.membership_id or "")
+    if row.owner_type == "user" and row.owner_user_id == user_id:
+        return True
+    if row.owner_type == "group" and row.owner_group_id in principal.group_ids:
+        return True
+    if reference.resource_type != "file":
+        return False
+    target_clauses = [
+        (FileShare.target_type == "user") & (FileShare.target_id == user_id),
+        (FileShare.target_type == "tenant")
+        & (FileShare.target_id == principal.tenant_id),
+    ]
+    if principal.group_ids:
+        target_clauses.append(
+            (FileShare.target_type == "group")
+            & (FileShare.target_id.in_(tuple(principal.group_ids)))
+        )
+    return session.scalar(
+        select(FileShare.id).where(
+            FileShare.tenant_id == principal.tenant_id,
+            FileShare.file_asset_id == row.id,
+            effective_file_share_clause(),
+            or_(*target_clauses),
+        ).limit(1)
+    ) is not None
+
+
+def _active_shares(session: Session, asset_id: str) -> Sequence[FileShare]:
+    return tuple(
+        session.scalars(
+            select(FileShare).where(
+                FileShare.file_asset_id == asset_id,
+                effective_file_share_clause(),
+            )
+        )
+    )
+
+
+def _shares_by_asset(
+    session: Session,
+    rows: Sequence[FileAsset | FileFolder],
+) -> dict[str, tuple[FileShare, ...]]:
+    asset_ids = [row.id for row in rows if isinstance(row, FileAsset)]
+    grouped: dict[str, list[FileShare]] = {asset_id: [] for asset_id in asset_ids}
+    if not asset_ids:
+        return {}
+    for share in session.scalars(
+        select(FileShare).where(
+            FileShare.file_asset_id.in_(asset_ids),
+            effective_file_share_clause(),
+        )
+    ):
+        grouped.setdefault(share.file_asset_id, []).append(share)
+    return {key: tuple(value) for key, value in grouped.items()}
+
+
+def _session(value: object) -> Session:
+    if not isinstance(value, Session):
+        raise TypeError("Files search requires a SQLAlchemy session.")
+    return value
+
+
+__all__ = [
+    "FilesSearchSource",
+    "PROVIDER_ID",
+    "create_files_search_source",
+]

src/govoplan_files/backend/storage/connector_browse.py

@@ -15,8 +15,8 @@ from defusedxml import ElementTree as SafeElementTree
 
 from govoplan_core.security.outbound_http import (
     OutboundHttpError,
-    validate_unpinned_sdk_host,
+    validate_outbound_host,
-    validate_unpinned_sdk_http_url,
+    validate_outbound_http_url,
 )
 
 from govoplan_files.backend.storage.http_client import ConnectorHttpError, request_connector_bytes
@@ -27,6 +27,12 @@ from govoplan_files.backend.storage.connector_deployment import (
     validate_connector_tls_metadata,
 )
 from govoplan_files.backend.storage.connector_profiles import ConnectorProfile
+from govoplan_files.backend.storage.sdk_peer_pinning import (
+    SdkPeerPinningError,
+    create_pinned_s3_client,
+    install_pinned_smb_transport,
+    pinned_smb_connection_cache,
+)
 
 
 class ConnectorBrowseError(RuntimeError):
@@ -246,6 +252,28 @@ def _browse_smb(profile: ConnectorProfile, *, path: str) -> list[ConnectorBrowse
 
 def _browse_s3(profile: ConnectorProfile, *, path: str, library_id: str | None, continuation_token: str | None) -> list[ConnectorBrowseItem]:
     client = _s3_client(profile)
+    try:
+        return _browse_s3_with_client(
+            client,
+            profile=profile,
+            path=path,
+            library_id=library_id,
+            continuation_token=continuation_token,
+        )
+    finally:
+        close = getattr(client, "close", None)
+        if callable(close):
+            close()
+
+
+def _browse_s3_with_client(
+    client: Any,
+    *,
+    profile: ConnectorProfile,
+    path: str,
+    library_id: str | None,
+    continuation_token: str | None,
+) -> list[ConnectorBrowseItem]:
     bucket = _s3_bucket(profile, library_id)
     if not bucket:
         try:
@@ -313,23 +341,22 @@ def _s3_client(profile: ConnectorProfile) -> Any:
         raise ConnectorBrowseError("Secret-ref S3 credentials need a runtime secret resolver before live browsing")
     if profile.endpoint_url:
         try:
-            endpoint_url = validate_unpinned_sdk_http_url(
+            endpoint_url = validate_outbound_http_url(
                 profile.endpoint_url,
                 label="S3 connector endpoint",
             )
         except OutboundHttpError as exc:
             raise ConnectorBrowseError(str(exc)) from exc
     else:
-        raise ConnectorBrowseError(
+        endpoint_url = None
-            "S3 connector endpoint discovery uses an SDK transport that cannot guarantee connection-time DNS/IP "
-            "pinning; live S3 access is disabled until that transport supports pinning"
-        )
     try:
-        boto3 = import_module("boto3")
         config_module = import_module("botocore.config")
+        unsigned = import_module("botocore").UNSIGNED
     except ImportError as exc:
         raise ConnectorBrowseUnsupported("S3 connector browsing requires the optional boto3 dependency") from exc
-    kwargs: dict[str, object] = {"endpoint_url": endpoint_url}
+    kwargs: dict[str, object] = {}
+    if endpoint_url:
+        kwargs["endpoint_url"] = endpoint_url
     region = _metadata_string(profile, "region") or _metadata_string(profile, "aws_region")
     if region:
         kwargs["region_name"] = region
@@ -346,10 +373,21 @@ def _s3_client(profile: ConnectorProfile) -> Any:
     if verify is not None:
         kwargs["verify"] = verify
     addressing_style = _s3_addressing_style(profile)
+    config_values: dict[str, object] = {
+        "proxies": {},
+        "retries": {"mode": "standard", "max_attempts": 4},
+    }
     if addressing_style:
-        kwargs["config"] = config_module.Config(s3={"addressing_style": addressing_style})
+        config_values["s3"] = {"addressing_style": addressing_style}
+    if bool(access_key) != bool(secret_key):
+        raise ConnectorBrowseError("S3 connectors require both an access key and a secret key")
+    if not access_key:
+        config_values["signature_version"] = unsigned
+    kwargs["config"] = config_module.Config(**config_values)
     try:
-        return boto3.client("s3", **kwargs)
+        return create_pinned_s3_client(**kwargs)
+    except SdkPeerPinningError as exc:
+        raise ConnectorBrowseError(str(exc)) from exc
     except Exception as exc:  # pragma: no cover - concrete exception types are dependency-version specific
         raise ConnectorBrowseError(f"S3 connector could not be initialized: {exc}") from exc
 
@@ -783,7 +821,7 @@ def _smb_location(profile: ConnectorProfile) -> _SmbLocation:
         raise ConnectorBrowseError("SMB connector endpoint_url must include a server")
     port = parsed.port or _int(profile.metadata.get("port")) or 445
     try:
-        validate_unpinned_sdk_host(server, port=port, label="SMB connector endpoint")
+        validate_outbound_host(server, port=port, label="SMB connector endpoint")
     except OutboundHttpError as exc:
         raise ConnectorBrowseError(str(exc)) from exc
     path_parts = [part for part in unquote(parsed.path or "").strip("/").split("/") if part]
@@ -810,6 +848,7 @@ def _smb_unc_path(location: _SmbLocation, path: str) -> str:
 def _smb_client_kwargs(profile: ConnectorProfile, location: _SmbLocation) -> dict[str, object]:
     kwargs: dict[str, object] = {
         "port": location.port,
+        "connection_cache": pinned_smb_connection_cache(),
         "require_signing": _metadata_bool(profile, "require_signing", default=True),
         "auth_protocol": _metadata_string(profile, "auth_protocol") or "ntlm",
     }
@@ -845,9 +884,11 @@ def _profile_token(profile: ConnectorProfile) -> str | None:
 
 def _smbclient_module() -> Any:
     try:
-        return import_module("smbclient")
+        return install_pinned_smb_transport(import_module("smbclient"))
     except ImportError as exc:
         raise ConnectorBrowseUnsupported("SMB connector browsing requires the optional smbprotocol dependency") from exc
+    except SdkPeerPinningError as exc:
+        raise ConnectorBrowseUnsupported(str(exc)) from exc
 
 
 def _smb_entry_stat(entry: object) -> object | None:

src/govoplan_files/backend/storage/connector_imports.py

@@ -314,15 +314,20 @@ def _read_s3_file(profile: ConnectorProfile, *, library_id: str, path: str, max_
     if not key:
         raise ConnectorImportError("S3 import requires an object key")
     client = _s3_import_client(profile)
-    detail = _s3_object_detail(client, bucket=bucket, key=key, max_bytes=max_bytes)
+    try:
-    version_id = _clean(detail.get("VersionId"))
+        detail = _s3_object_detail(client, bucket=bucket, key=key, max_bytes=max_bytes)
-    response, data = _download_s3_object(
+        version_id = _clean(detail.get("VersionId"))
-        client,
+        response, data = _download_s3_object(
-        bucket=bucket,
+            client,
-        key=key,
+            bucket=bucket,
-        version_id=version_id,
+            key=key,
-        max_bytes=max_bytes,
+            version_id=version_id,
-    )
+            max_bytes=max_bytes,
+        )
+    finally:
+        close = getattr(client, "close", None)
+        if callable(close):
+            close()
     content_type = _clean(response.get("ContentType") if isinstance(response, dict) else None) or _clean(detail.get("ContentType")) or mimetypes.guess_type(key)[0]
     etag = _clean(response.get("ETag") if isinstance(response, dict) else None) or _clean(detail.get("ETag"))
     filename = filename_from_path(key)

src/govoplan_files/backend/storage/connector_providers.py

@@ -110,7 +110,7 @@ def connector_provider_descriptors() -> tuple[ConnectorProviderDescriptor, ...]:
             conflict_strategy="Managed import/sync uses existing files conflict_strategy handling after download.",
             preview_strategy="Previews are generated from the frozen managed file after import or sync, not directly from the share.",
             audit_events=("files.connector.imported", "files.connector.synced", "files.connector.accessed"),
-            notes="Browse/import logic is implemented, but live smbprotocol access fails closed until the SDK supports connection-time DNS/IP pinning for initial connections and DFS referrals.",
+            notes="Initial sessions, reconnects, aliases, and DFS referral targets use the Files-owned pinned smbprotocol transport and the deployment-wide private-network policy.",
         ),
         ConnectorProviderDescriptor(
             provider="s3",
@@ -125,7 +125,7 @@ def connector_provider_descriptors() -> tuple[ConnectorProviderDescriptor, ...]:
             conflict_strategy="Managed import/sync uses existing files conflict_strategy handling after object download.",
             preview_strategy="Previews are generated from the frozen managed file after import or sync, not directly from the bucket.",
             audit_events=("files.connector.imported", "files.connector.synced", "files.connector.accessed"),
-            notes="Browse/import logic is implemented, but live boto3 access fails closed until its HTTP transport supports connection-time DNS/IP pinning and redirect revalidation.",
+            notes="Every botocore connection pool uses the Files-owned pinned transport, including retries, redirects, endpoint discovery, and virtual-host aliases; outbound proxies and ambient credential discovery are disabled.",
         ),
         ConnectorProviderDescriptor(
             provider="sharepoint",

src/govoplan_files/backend/storage/connector_visibility.py

@@ -136,11 +136,6 @@ def connector_profile_usable_for_import(profile: ConnectorProfile) -> bool:
         and descriptor.import_supported
     ):
         return False
-    # These SDK paths intentionally fail closed until every connection peer and
-    # SDK-managed redirect/referral can be pinned and revalidated.
-    if profile.provider in {"s3", "smb"}:
-        return False
-
     # Prove that the initial root browse performed by the current Files UI is
     # policy-allowed. A later selected remote path/item is checked again.
     return connector_policy_decision(

src/govoplan_files/backend/storage/files.py

@@ -596,21 +596,31 @@ def _asset_visibility_query_for_user(
         if owner_type == "group" and owner_id:
             query = query.filter(FileAsset.owner_group_id == owner_id)
     if campaign_id:
-        query = query.join(FileShare, FileShare.file_asset_id == FileAsset.id).filter(
+        campaign_share = exists().where(
             FileShare.tenant_id == tenant_id,
+            FileShare.file_asset_id == FileAsset.id,
             FileShare.target_type == "campaign",
             FileShare.target_id == campaign_id,
             effective_file_share_clause(),
         )
+        query = query.filter(campaign_share)
     elif not is_admin and not owner_type:
         group_ids = user_group_ids(session, tenant_id=tenant_id, user_id=user_id)
-        query = query.outerjoin(FileShare, FileShare.file_asset_id == FileAsset.id).filter(
+        active_share = exists().where(
+            FileShare.tenant_id == tenant_id,
+            FileShare.file_asset_id == FileAsset.id,
+            effective_file_share_clause(),
+            or_(
+                (FileShare.target_type == "user") & (FileShare.target_id == user_id),
+                (FileShare.target_type == "group") & (FileShare.target_id.in_(group_ids)),
+                (FileShare.target_type == "tenant") & (FileShare.target_id == tenant_id),
+            )
+        )
+        query = query.filter(
             or_(
                 (FileAsset.owner_type == "user") & (FileAsset.owner_user_id == user_id),
                 (FileAsset.owner_type == "group") & (FileAsset.owner_group_id.in_(group_ids)),
-                effective_file_share_clause() & (FileShare.target_type == "user") & (FileShare.target_id == user_id),
+                active_share,
-                effective_file_share_clause() & (FileShare.target_type == "group") & (FileShare.target_id.in_(group_ids)),
-                effective_file_share_clause() & (FileShare.target_type == "tenant") & (FileShare.target_id == tenant_id),
             )
         )
     if path_prefix:
@@ -639,7 +649,7 @@ def _asset_visibility_query_for_user(
             CampaignAttachmentUse.use_stage == "sent",
         )
         query = query.filter(sent_use_exists if audit_relevant else ~sent_use_exists)
-    return query.distinct()
+    return query
 
 
 def count_assets_for_user(

src/govoplan_files/backend/storage/integrity.py

@@ -99,6 +99,7 @@ def run_integrity_scan_batch(
         scan.phase = "completed"
         scan.status = "completed"
         scan.completed_at = utcnow()
+    scan.revision += 1
     session.add(scan)
     return scan
 
@@ -110,6 +111,7 @@ def mark_integrity_scan_failed(
 ) -> None:
     scan.status = "failed"
     scan.last_error = type(error).__name__[:255]
+    scan.revision += 1
 
 
 def inspect_blob(
@@ -269,6 +271,7 @@ def recheck_integrity_finding(
             finding.resolved_at = utcnow()
             finding.resolved_by_user_id = user_id
         session.add(finding)
+        finding.revision += 1
         changed = previous != (
             blob.integrity_status,
             blob.quarantined_at,
@@ -312,6 +315,7 @@ def cleanup_orphan_finding(
             finding.state = "resolved"
             finding.resolved_at = utcnow()
             finding.resolved_by_user_id = user_id
+            finding.revision += 1
             session.add(finding)
             return IntegrityActionResult(
                 action="retained_referenced",
@@ -360,6 +364,7 @@ def cleanup_orphan_finding(
     finding.state = "deleted"
     finding.resolved_at = utcnow()
     finding.resolved_by_user_id = user_id
+    finding.revision += 1
     session.add(finding)
     return IntegrityActionResult(
         action=action,
@@ -465,6 +470,8 @@ def _record_blob_finding(
             expected_size_bytes=blob.storage_size_bytes if blob.storage_size_bytes is not None else blob.size_bytes,
             expected_checksum_sha256=blob.storage_checksum_sha256 if blob.storage_checksum_sha256 is not None else blob.checksum_sha256,
         )
+    else:
+        finding.revision += 1
     _update_finding_from_inspection(finding, inspection)
     session.add(finding)
     return finding
@@ -514,6 +521,7 @@ def _resolve_scan_blob_findings(
     ):
         finding.state = "resolved"
         finding.resolved_at = utcnow()
+        finding.revision += 1
         session.add(finding)
 
 

src/govoplan_files/backend/storage/sdk_peer_pinning.py

@@ -0,0 +1,331 @@
+from __future__ import annotations
+
+import inspect
+import socket
+import threading
+from functools import lru_cache
+from importlib import import_module
+from typing import Any
+
+from govoplan_core.security.outbound_http import (
+    OutboundHttpError,
+    create_outbound_connection,
+)
+
+
+class SdkPeerPinningError(RuntimeError):
+    """Raised when an optional SDK cannot be bound to the pinned transport."""
+
+
+_BOTOCORE_CLIENT_CREATION_LOCK = threading.Lock()
+
+
+def create_pinned_s3_client(**kwargs: Any) -> Any:
+    """Construct an S3 client whose first and every later socket is pinned.
+
+    Botocore does not expose the HTTP-session class through boto3's public
+    client API. Its endpoint creator does expose that seam, so this function
+    replaces the creator only for the bounded client-construction operation.
+    The lock avoids an unsafe interleaving with another GovOPlaN client
+    construction. Any unrelated botocore client created during the short
+    replacement window also receives the stricter transport.
+    """
+
+    try:
+        boto3 = import_module("boto3")
+        botocore_args = import_module("botocore.args")
+    except ImportError as exc:
+        raise SdkPeerPinningError(
+            "S3 connector browsing requires the optional boto3 dependency"
+        ) from exc
+
+    pinned_session_cls = _botocore_transport_types()[0]
+    original_creator = getattr(botocore_args, "EndpointCreator", None)
+    if original_creator is None or not hasattr(original_creator, "create_endpoint"):
+        raise SdkPeerPinningError(
+            "The installed botocore release does not expose the endpoint transport seam required for peer pinning"
+        )
+
+    class PinnedEndpointCreator(original_creator):  # type: ignore[misc, valid-type]
+        def create_endpoint(self, *args: Any, **endpoint_kwargs: Any) -> Any:
+            endpoint_kwargs["http_session_cls"] = pinned_session_cls
+            return super().create_endpoint(*args, **endpoint_kwargs)
+
+    with _BOTOCORE_CLIENT_CREATION_LOCK:
+        current_creator = getattr(botocore_args, "EndpointCreator", None)
+        if current_creator is not original_creator:
+            raise SdkPeerPinningError(
+                "Botocore endpoint construction changed concurrently; refusing to create an unproven S3 client"
+            )
+        botocore_args.EndpointCreator = PinnedEndpointCreator
+        try:
+            session_type = getattr(getattr(boto3, "session", None), "Session", None)
+            if session_type is None:
+                raise SdkPeerPinningError(
+                    "The installed boto3 release does not expose its isolated session constructor"
+                )
+            client = session_type().client("s3", **kwargs)
+        finally:
+            if getattr(botocore_args, "EndpointCreator", None) is PinnedEndpointCreator:
+                botocore_args.EndpointCreator = original_creator
+
+    http_session = getattr(getattr(client, "_endpoint", None), "http_session", None)
+    if http_session is None or not isinstance(http_session, pinned_session_cls):
+        close = getattr(client, "close", None)
+        if callable(close):
+            close()
+        raise SdkPeerPinningError(
+            "Botocore did not install the required pinned HTTP transport; the S3 client was discarded"
+        )
+    return client
+
+
+@lru_cache(maxsize=1)
+def _botocore_transport_types() -> tuple[type[Any], type[Any], type[Any]]:
+    try:
+        awsrequest = import_module("botocore.awsrequest")
+        httpsession = import_module("botocore.httpsession")
+        urllib3_exceptions = import_module("urllib3.exceptions")
+    except ImportError as exc:
+        raise SdkPeerPinningError(
+            "S3 connector browsing requires a compatible botocore HTTP transport"
+        ) from exc
+
+    required = (
+        "AWSHTTPConnection",
+        "AWSHTTPSConnection",
+        "AWSHTTPConnectionPool",
+        "AWSHTTPSConnectionPool",
+    )
+    if any(not hasattr(awsrequest, name) for name in required) or not hasattr(
+        httpsession, "URLLib3Session"
+    ):
+        raise SdkPeerPinningError(
+            "The installed botocore release is missing the connection classes required for peer pinning"
+        )
+
+    def pinned_new_connection(connection: Any) -> socket.socket:
+        hostname = str(getattr(connection, "_dns_host", "") or "").strip()
+        port = int(getattr(connection, "port", 0) or 0)
+        if not hostname or not port:
+            raise urllib3_exceptions.NewConnectionError(
+                connection, "Pinned S3 connection is missing its target authority"
+            )
+        try:
+            return create_outbound_connection(
+                hostname,
+                port,
+                timeout=getattr(connection, "timeout", None),
+                source_address=getattr(connection, "source_address", None),
+                socket_options=getattr(connection, "socket_options", None),
+                label="S3 connector peer",
+            )
+        except socket.timeout as exc:
+            raise urllib3_exceptions.ConnectTimeoutError(
+                connection,
+                f"Connection to {hostname} timed out while selecting an approved peer",
+            ) from exc
+        except (OSError, OutboundHttpError, ValueError) as exc:
+            raise urllib3_exceptions.NewConnectionError(
+                connection,
+                f"S3 connector peer was rejected: {exc}",
+            ) from exc
+
+    class PinnedAWSHTTPConnection(awsrequest.AWSHTTPConnection):
+        _new_conn = pinned_new_connection
+
+    class PinnedAWSHTTPSConnection(awsrequest.AWSHTTPSConnection):
+        _new_conn = pinned_new_connection
+
+    class PinnedAWSHTTPConnectionPool(awsrequest.AWSHTTPConnectionPool):
+        ConnectionCls = PinnedAWSHTTPConnection
+
+    class PinnedAWSHTTPSConnectionPool(awsrequest.AWSHTTPSConnectionPool):
+        ConnectionCls = PinnedAWSHTTPSConnection
+
+    class PinnedURLLib3Session(httpsession.URLLib3Session):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            # A proxy would select the final target outside this process. Until
+            # proxy peer delegation is modeled, connector traffic is direct.
+            kwargs["proxies"] = {}
+            super().__init__(*args, **kwargs)
+            self._pool_classes_by_scheme = {
+                "http": PinnedAWSHTTPConnectionPool,
+                "https": PinnedAWSHTTPSConnectionPool,
+            }
+            manager = getattr(self, "_manager", None)
+            if manager is None or not hasattr(manager, "pool_classes_by_scheme"):
+                raise SdkPeerPinningError(
+                    "The installed botocore pool manager cannot enforce pinned connection classes"
+                )
+            manager.pool_classes_by_scheme = self._pool_classes_by_scheme
+
+    return PinnedURLLib3Session, PinnedAWSHTTPConnection, PinnedAWSHTTPSConnection
+
+
+@lru_cache(maxsize=1)
+def pinned_smb_connection_cache() -> dict[str, Any]:
+    """Return the Files-owned cache; unproven process-global sessions are never reused."""
+
+    return {}
+
+
+def install_pinned_smb_transport(smbclient: Any) -> Any:
+    """Install a process-wide fail-closed smbclient session factory.
+
+    smbclient routes initial connections, reconnects, and DFS targets through
+    ``smbclient._pool.register_session``. Replacing that single factory with a
+    behavior-compatible implementation ensures every target uses a socket
+    selected by Core's deployment-wide outbound policy.
+    """
+
+    try:
+        pool = import_module("smbclient._pool")
+        connection_module = import_module("smbprotocol.connection")
+        session_module = import_module("smbprotocol.session")
+        transport_module = import_module("smbprotocol.transport")
+    except ImportError as exc:
+        raise SdkPeerPinningError(
+            "SMB connector browsing requires the optional smbprotocol dependency"
+        ) from exc
+
+    current = getattr(pool, "register_session", None)
+    if getattr(current, "__govoplan_peer_pinned__", False):
+        expected_tcp = getattr(current, "__govoplan_pinned_tcp__", None)
+        if expected_tcp is None or getattr(connection_module, "Tcp", None) is not expected_tcp:
+            raise SdkPeerPinningError(
+                "The installed SMB transport changed after peer pinning; refusing to reuse the session factory"
+            )
+        return smbclient
+    required_parameters = {
+        "server",
+        "username",
+        "password",
+        "port",
+        "encrypt",
+        "connection_timeout",
+        "connection_cache",
+        "auth_protocol",
+        "require_signing",
+    }
+    if current is None or not required_parameters.issubset(
+        inspect.signature(current).parameters
+    ):
+        raise SdkPeerPinningError(
+            "The installed smbprotocol release does not expose the session seam required for peer pinning"
+        )
+
+    class PinnedTcp(transport_module.Tcp):
+        def connect(self) -> None:
+            with self._sock_lock:
+                if self.connected:
+                    return
+                try:
+                    self._sock = create_outbound_connection(
+                        self.server,
+                        int(self.port),
+                        timeout=self.timeout,
+                        label="SMB connector peer",
+                    )
+                except (OSError, OutboundHttpError, ValueError) as exc:
+                    raise ValueError(
+                        f"SMB connector peer '{self.server}:{self.port}' was rejected: {exc}"
+                    ) from exc
+                self._sock.settimeout(None)
+                self.connected = True
+
+    # Connection.connect() instantiates the module-level Tcp symbol on every
+    # reconnect. Replacing it process-wide is deliberate: an SMB connection
+    # created by another code path must become stricter, never bypass Files'
+    # peer boundary.
+    if not hasattr(connection_module, "Tcp"):
+        raise SdkPeerPinningError(
+            "The installed smbprotocol release cannot install the pinned TCP transport"
+        )
+    connection_module.Tcp = PinnedTcp
+
+    def pinned_register_session(
+        server: str,
+        username: str | None = None,
+        password: str | None = None,
+        port: int = 445,
+        encrypt: bool | None = None,
+        connection_timeout: float = 60,
+        connection_cache: dict[str, Any] | None = None,
+        auth_protocol: str = "negotiate",
+        require_signing: bool = True,
+    ) -> Any:
+        cache = pinned_smb_connection_cache() if connection_cache is None else connection_cache
+        connection_key = f"{server.lower()}:{port}"
+        connection = cache.get(connection_key)
+        transport = getattr(connection, "transport", None)
+        if connection is not None and not isinstance(transport, PinnedTcp):
+            disconnect = getattr(connection, "disconnect", None)
+            if callable(disconnect):
+                try:
+                    disconnect(close=True)
+                except Exception:
+                    pass
+            cache.pop(connection_key, None)
+            connection = None
+        if connection is None or not getattr(connection.transport, "connected", False):
+            connection = connection_module.Connection(
+                pool.ClientConfig().client_guid,
+                server,
+                port,
+                require_signing=require_signing,
+            )
+            connection.transport = PinnedTcp(server, port)
+            connection.connect(timeout=connection_timeout)
+            if not isinstance(connection.transport, PinnedTcp):
+                disconnect = getattr(connection, "disconnect", None)
+                if callable(disconnect):
+                    try:
+                        disconnect(close=True)
+                    except Exception:
+                        pass
+                raise SdkPeerPinningError(
+                    "smbprotocol replaced the required pinned TCP transport during connection setup"
+                )
+            cache[connection_key] = connection
+
+        session = next(
+            (
+                item
+                for item in connection.session_table.values()
+                if username is None or item.username == username
+            ),
+            None,
+        )
+        if session is None:
+            session = session_module.Session(
+                connection,
+                username=username,
+                password=password,
+                require_encryption=(encrypt is True),
+                auth_protocol=auth_protocol,
+            )
+            session.connect()
+        elif encrypt is not None:
+            if session.encrypt_data and not encrypt:
+                raise ValueError(
+                    "Cannot disable encryption on an already negotiated session."
+                )
+            if not session.encrypt_data and encrypt:
+                session.encrypt = True
+        return session
+
+    pinned_register_session.__govoplan_peer_pinned__ = True  # type: ignore[attr-defined]
+    pinned_register_session.__govoplan_pinned_tcp__ = PinnedTcp  # type: ignore[attr-defined]
+    pool.register_session = pinned_register_session
+    if hasattr(smbclient, "register_session"):
+        smbclient.register_session = pinned_register_session
+    return smbclient
+
+
+__all__ = [
+    "SdkPeerPinningError",
+    "create_pinned_s3_client",
+    "install_pinned_smb_transport",
+    "pinned_smb_connection_cache",
+]

tests/test_connector_providers.py

@@ -5,7 +5,7 @@ from datetime import UTC, datetime
 from unittest.mock import patch
 
 from govoplan_files.backend.storage.connector_browse import ConnectorBrowseError, _smb_location, browse_connector_profile
-from govoplan_files.backend.storage.connector_imports import ConnectorImportError, read_connector_file
+from govoplan_files.backend.storage.connector_imports import read_connector_file
 from govoplan_files.backend.storage.connector_profiles import ConnectorProfile, connector_profiles_from_payload
 from govoplan_files.backend.storage.connector_providers import connector_provider_descriptors
 
@@ -23,6 +23,10 @@ class FakeS3Client:
         self.list_objects_request: dict[str, object] | None = None
         self.head_request: dict[str, object] | None = None
         self.get_request: dict[str, object] | None = None
+        self.closed = False
+
+    def close(self) -> None:
+        self.closed = True
 
     def list_objects_v2(self, **kwargs: object) -> dict[str, object]:
         self.list_objects_request = dict(kwargs)
@@ -75,42 +79,39 @@ def s3_profile(**overrides: object) -> ConnectorProfile:
 
 
 class ConnectorProviderTests(unittest.TestCase):
-    def test_smb_sdk_transport_fails_closed_before_client_creation_in_all_modes(self) -> None:
+    def test_smb_browse_uses_the_files_owned_pinned_connection_cache(self) -> None:
         profile = ConnectorProfile(
             id="smb",
             label="SMB",
             provider="smb",
             endpoint_url="smb://files.example.test/share",
         )
-        for allow_private, address in ((False, "93.184.216.34"), (True, "10.0.0.5")):
+        with patch.dict(
-            with self.subTest(allow_private=allow_private), patch.dict(
+            "os.environ",
-                "os.environ",
+            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
-                {
+        ), patch(
-                    "APP_ENV": "production",
+            "govoplan_core.security.outbound_http.socket.getaddrinfo",
-                    "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": str(allow_private).lower(),
+            return_value=[(2, 1, 6, "", ("93.184.216.34", 445))],
-                },
+        ), patch("govoplan_files.backend.storage.connector_browse._smbclient_module") as sdk:
-            ), patch(
+            sdk.return_value.scandir.return_value.__enter__.return_value = iter(())
-                "govoplan_core.security.outbound_http.socket.getaddrinfo",
+            self.assertEqual([], browse_connector_profile(profile, path=""))
-                return_value=[(2, 1, 6, "", (address, 445))],
-            ), patch("govoplan_files.backend.storage.connector_browse._smbclient_module") as sdk, self.assertRaisesRegex(
-                ConnectorBrowseError,
-                "redirects/referrals.*DNS/IP pinning",
-            ):
-                browse_connector_profile(profile, path="")
-            sdk.assert_not_called()
 
-    def test_smb_explicit_ip_still_fails_closed_because_the_sdk_may_follow_referrals(self) -> None:
+        kwargs = sdk.return_value.scandir.call_args.kwargs
+        self.assertIsInstance(kwargs["connection_cache"], dict)
+        self.assertTrue(kwargs["require_signing"])
+
+    def test_smb_endpoint_preflight_applies_private_network_policy(self) -> None:
         profile = ConnectorProfile(id="smb", label="SMB", provider="smb", endpoint_url="smb://10.0.0.5/share")
         with patch.dict(
             "os.environ",
-            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"},
+            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
         ), patch(
             "govoplan_core.security.outbound_http.socket.getaddrinfo",
             return_value=[(2, 1, 6, "", ("10.0.0.5", 445))],
-        ), self.assertRaisesRegex(ConnectorBrowseError, "redirects/referrals.*DNS/IP pinning"):
+        ), self.assertRaisesRegex(ConnectorBrowseError, "non-public network"):
             _smb_location(profile)
 
-    def test_smb_import_surfaces_fail_closed_policy_as_an_import_error(self) -> None:
+    def test_smb_import_uses_the_same_pinned_connection_cache(self) -> None:
         profile = ConnectorProfile(id="smb", label="SMB", provider="smb", endpoint_url="smb://10.0.0.5/share")
         with patch.dict(
             "os.environ",
@@ -118,12 +119,13 @@ class ConnectorProviderTests(unittest.TestCase):
         ), patch(
             "govoplan_core.security.outbound_http.socket.getaddrinfo",
             return_value=[(2, 1, 6, "", ("10.0.0.5", 445))],
-        ), patch("govoplan_files.backend.storage.connector_imports._smbclient_module") as sdk, self.assertRaisesRegex(
+        ), patch("govoplan_files.backend.storage.connector_imports._smbclient_module") as sdk:
-            ConnectorImportError,
+            sdk.return_value.stat.return_value.st_size = 4
-            "redirects/referrals.*DNS/IP pinning",
+            sdk.return_value.open_file.return_value.__enter__.return_value.read.return_value = b"test"
-        ):
+            downloaded = read_connector_file(profile, library_id="", path="notice.txt", max_bytes=1024)
-            read_connector_file(profile, library_id="", path="notice.txt", max_bytes=1024)
+
-        sdk.assert_not_called()
+        self.assertEqual(b"test", downloaded.data)
+        self.assertIsInstance(sdk.return_value.stat.call_args.kwargs["connection_cache"], dict)
 
     def test_provider_descriptors_include_s3_and_reserved_microsoft_providers(self) -> None:
         descriptors = {descriptor.provider: descriptor for descriptor in connector_provider_descriptors()}
@@ -172,6 +174,7 @@ class ConnectorProviderTests(unittest.TestCase):
         self.assertEqual("govoplan:root/report.xlsx", items[1].external_id)
         self.assertEqual("root/report.xlsx", items[1].metadata["key"])
         self.assertEqual("next-page", items[1].metadata["next_continuation_token"])
+        self.assertTrue(client.closed)
 
     def test_s3_browse_lists_buckets_when_profile_has_no_bucket(self) -> None:
         client = FakeS3Client()
@@ -181,27 +184,33 @@ class ConnectorProviderTests(unittest.TestCase):
 
         self.assertEqual(["archive"], [item.path for item in items])
 
-    def test_s3_sdk_transport_fails_closed_before_client_creation_in_private_mode(self) -> None:
+    def test_s3_client_is_constructed_through_the_pinned_transport(self) -> None:
         with patch.dict(
             "os.environ",
             {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"},
         ), patch(
             "govoplan_core.security.outbound_http.socket.getaddrinfo",
             return_value=[(2, 1, 6, "", ("127.0.0.1", 9000))],
-        ), patch("govoplan_files.backend.storage.connector_browse.import_module") as importer, self.assertRaisesRegex(
+        ), patch(
-            ConnectorBrowseError,
+            "govoplan_files.backend.storage.connector_browse.create_pinned_s3_client",
-            "until that transport supports.*DNS/IP pinning",
+            return_value=FakeS3Client(),
-        ):
+        ) as factory:
             browse_connector_profile(s3_profile(), path="")
-        importer.assert_not_called()
 
-    def test_s3_sdk_endpoint_discovery_fails_closed(self) -> None:
+        kwargs = factory.call_args.kwargs
-        with patch("govoplan_files.backend.storage.connector_browse.import_module") as importer, self.assertRaisesRegex(
+        self.assertEqual("http://127.0.0.1:9000", kwargs["endpoint_url"])
-            ConnectorBrowseError,
+        self.assertEqual("access-key", kwargs["aws_access_key_id"])
-            "endpoint discovery.*cannot guarantee.*DNS/IP pinning",
+        self.assertEqual("secret-key", kwargs["aws_secret_access_key"])
-        ):
+        self.assertEqual({}, kwargs["config"].proxies)
+
+    def test_s3_endpoint_discovery_uses_the_same_pinned_transport(self) -> None:
+        with patch(
+            "govoplan_files.backend.storage.connector_browse.create_pinned_s3_client",
+            return_value=FakeS3Client(),
+        ) as factory:
             browse_connector_profile(s3_profile(endpoint_url=None), path="")
-        importer.assert_not_called()
+
+        self.assertNotIn("endpoint_url", factory.call_args.kwargs)
 
     def test_s3_import_downloads_object_and_preserves_remote_identity(self) -> None:
         client = FakeS3Client()
@@ -217,6 +226,7 @@ class ConnectorProviderTests(unittest.TestCase):
         self.assertEqual("govoplan:root/report.txt", downloaded.external_id)
         self.assertEqual("s3://govoplan/root/report.txt", downloaded.external_url)
         self.assertEqual("checksum", downloaded.metadata["checksum_sha256"])
+        self.assertTrue(client.closed)
 
 
 if __name__ == "__main__":

tests/test_connector_visibility.py

@@ -10,6 +10,9 @@ from govoplan_files.backend.storage.connector_visibility import (
     connector_profile_usable_for_import,
     visible_connector_profiles_for_actor,
 )
+from govoplan_files.backend.storage.connector_providers import (
+    connector_provider_descriptors,
+)
 
 
 def _profile(
@@ -168,12 +171,16 @@ class ConnectorVisibilityTests(unittest.TestCase):
         self,
     ) -> None:
         self.assertTrue(connector_profile_usable_for_import(_profile("webdav")))
-        self.assertFalse(
+        descriptors = {
-            connector_profile_usable_for_import(_profile("s3", provider="s3"))
+            item.provider: item for item in connector_provider_descriptors()
-        )
+        }
-        self.assertFalse(
+        for provider in ("s3", "smb"):
-            connector_profile_usable_for_import(_profile("smb", provider="smb"))
+            self.assertEqual(
-        )
+                descriptors[provider].installed,
+                connector_profile_usable_for_import(
+                    _profile(provider, provider=provider)
+                ),
+            )
         self.assertFalse(
             connector_profile_usable_for_import(
                 _profile("sharepoint", provider="sharepoint")

tests/test_documentation.py

@@ -142,6 +142,7 @@ class FilesRuntimeDocumentationTests(unittest.TestCase):
             base_path="/classified",
             credential_mode="basic",
             password_value="credential-secret",
+            secret_ref="runtime/connector-secret",
         )
         with patch(
             "govoplan_files.backend.documentation.visible_connector_profiles_for_actor",

tests/test_integrity_reconciliation.py

@@ -6,6 +6,7 @@ import unittest
 from pathlib import Path
 from unittest.mock import patch
 
+from fastapi import HTTPException
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
@@ -16,6 +17,7 @@ from govoplan_files.backend.db.models import (
     FileIntegrityFinding,
     FileIntegrityScan,
 )
+from govoplan_files.backend.routes.integrity import _assert_expected_revision
 from govoplan_files.backend.storage.backends import LocalFilesystemStorageBackend
 from govoplan_files.backend.storage.common import FileStorageError
 from govoplan_files.backend.storage.integrity import (
@@ -82,6 +84,7 @@ class IntegrityReconciliationTests(unittest.TestCase):
             backend=self.backend,
         )
         self.session.commit()
+        self.assertEqual(1, scan.revision)
 
         invocations = 0
         while scan.status != "completed":
@@ -99,6 +102,7 @@ class IntegrityReconciliationTests(unittest.TestCase):
                 self.fail("Integrity scan did not complete")
 
         self.assertGreater(invocations, 3)
+        self.assertEqual(1 + invocations, scan.revision)
         self.assertEqual(3, scan.scanned_blob_count)
         self.assertEqual(1, scan.verified_blob_count)
         self.assertEqual(2, scan.quarantined_blob_count)
@@ -183,6 +187,13 @@ class IntegrityReconciliationTests(unittest.TestCase):
         self.assertEqual("already_deleted", repeated.action)
         self.assertFalse(repeated.changed)
 
+    def test_stale_integrity_action_revision_is_rejected(self) -> None:
+        with self.assertRaises(HTTPException) as captured:
+            _assert_expected_revision(3, 2)
+
+        self.assertEqual(409, captured.exception.status_code)
+        self.assertIn("reload", str(captured.exception.detail).lower())
+
 
 def _blob(blob_id: str, filename: str, expected_data: bytes) -> FileBlob:
     return FileBlob(

tests/test_manifest_documentation.py

@@ -4,6 +4,7 @@ import unittest
 
 
 STATIC_TOPIC_IDS = {
+    "files.search.managed-content",
     "files.workflow.organize-managed-files",
     "files.workflow.find-and-download-files",
     "files.workflow.share-managed-files",
@@ -152,7 +153,7 @@ class FilesManifestDocumentationTests(unittest.TestCase):
             "/api/v1/files/connectors/credentials", {link.href for link in topic.links}
         )
 
-    def test_operator_topic_covers_recovery_and_fail_closed_s3_smb(self) -> None:
+    def test_operator_topic_covers_recovery_and_pinned_s3_smb(self) -> None:
         topic = self.topic(
             "files.reference.integrity-recovery-and-fail-closed-transports"
         )
@@ -163,6 +164,7 @@ class FilesManifestDocumentationTests(unittest.TestCase):
         self.assertIn("SMB", topic.body)
         self.assertIn("fail closed", topic.body)
         self.assertIn("DFS referrals", topic.body)
+        self.assertIn("ambient credential discovery", topic.body)
         self.assertIn("does not remove backend blob objects", topic.body)
         self.assertIn("bounded resumable integrity scan", topic.body)
         self.assertIn("quarantined", topic.body)
@@ -175,6 +177,14 @@ class FilesManifestDocumentationTests(unittest.TestCase):
             "/api/v1/files/integrity/scans",
             {link.href for link in topic.links},
         )
+        self.assertIn(
+            "/admin?section=tenant-file-integrity",
+            {link.href for link in topic.links},
+        )
+        self.assertIn(
+            "files.admin.tenant-integrity",
+            {surface.id for surface in self.manifest.frontend.view_surfaces},
+        )
         self.assertIn(
             "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS", topic.configuration_keys
         )

tests/test_sdk_peer_pinning.py

@@ -0,0 +1,294 @@
+from __future__ import annotations
+
+import socket
+import sys
+import threading
+import types
+import unittest
+from importlib.util import find_spec
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from unittest.mock import patch
+
+from govoplan_core.security.outbound_http import (
+    OutboundHttpBlocked,
+    create_outbound_connection as core_create_outbound_connection,
+)
+from govoplan_files.backend.storage.sdk_peer_pinning import (
+    SdkPeerPinningError,
+    _botocore_transport_types,
+    create_pinned_s3_client,
+    install_pinned_smb_transport,
+    pinned_smb_connection_cache,
+)
+
+
+class _Socket:
+    def __init__(self) -> None:
+        self.connected_to: object | None = None
+
+    def settimeout(self, _value: object) -> None:
+        return None
+
+    def setsockopt(self, *_args: object) -> None:
+        return None
+
+    def connect(self, value: object) -> None:
+        self.connected_to = value
+
+    def close(self) -> None:
+        return None
+
+
+class _S3Handler(BaseHTTPRequestHandler):
+    def do_GET(self) -> None:  # noqa: N802 - stdlib handler contract
+        body = (
+            b'<?xml version="1.0" encoding="UTF-8"?>'
+            b'<ListAllMyBucketsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">'
+            b'<Owner><ID>govoplan</ID><DisplayName>GovOPlaN</DisplayName></Owner>'
+            b'<Buckets><Bucket><Name>evidence</Name><CreationDate>2026-08-04T00:00:00Z</CreationDate>'
+            b'</Bucket></Buckets></ListAllMyBucketsResult>'
+        )
+        self.send_response(200)
+        self.send_header("Content-Type", "application/xml")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, _format: str, *_args: object) -> None:
+        return None
+
+
+@unittest.skipUnless(find_spec("botocore") and find_spec("boto3"), "boto3 extra is not installed")
+class S3PeerPinningTests(unittest.TestCase):
+    def test_client_is_born_with_the_pinned_http_session(self) -> None:
+        client = create_pinned_s3_client(
+            endpoint_url="http://127.0.0.1:9000",
+            region_name="eu-central-1",
+            aws_access_key_id="access",
+            aws_secret_access_key="secret",
+        )
+        try:
+            session_type, _, _ = _botocore_transport_types()
+            self.assertIsInstance(client._endpoint.http_session, session_type)
+            self.assertEqual({}, client._endpoint.http_session._proxy_config._proxies)
+        finally:
+            client.close()
+
+    def test_each_connection_attempt_resolves_and_pins_again(self) -> None:
+        _, http_connection, _ = _botocore_transport_types()
+        connection = http_connection(host="objects.example.test", port=80)
+        sockets = [_Socket(), _Socket()]
+        with patch(
+            "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+            side_effect=sockets,
+        ) as connector:
+            self.assertIs(sockets[0], connection._new_conn())
+            self.assertIs(sockets[1], connection._new_conn())
+
+        self.assertEqual(2, connector.call_count)
+        self.assertEqual("objects.example.test", connector.call_args.args[0])
+
+    def test_real_sdk_request_uses_the_pinned_socket_path(self) -> None:
+        server = ThreadingHTTPServer(("127.0.0.1", 0), _S3Handler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        endpoint = f"http://127.0.0.1:{server.server_port}"
+        try:
+            with patch.dict(
+                "os.environ",
+                {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "true"},
+            ), patch(
+                "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+                wraps=core_create_outbound_connection,
+            ) as connector:
+                client = create_pinned_s3_client(
+                    endpoint_url=endpoint,
+                    region_name="eu-central-1",
+                    aws_access_key_id="access",
+                    aws_secret_access_key="secret",
+                )
+                try:
+                    response = client.list_buckets()
+                finally:
+                    client.close()
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join(timeout=2)
+
+        self.assertEqual("evidence", response["Buckets"][0]["Name"])
+        self.assertEqual("127.0.0.1", connector.call_args.args[0])
+
+    def test_every_sdk_selected_origin_uses_a_pinned_pool(self) -> None:
+        session_type, http_connection, https_connection = _botocore_transport_types()
+        session = session_type(proxies={})
+        try:
+            first = session._manager.connection_from_url("http://one.example.test/root")
+            redirected = session._manager.connection_from_url("https://two.example.test/root")
+            self.assertIs(first.ConnectionCls, http_connection)
+            self.assertIs(redirected.ConnectionCls, https_connection)
+        finally:
+            session.close()
+
+    def test_tls_connection_retains_the_configured_authority_for_sni(self) -> None:
+        _, _, https_connection = _botocore_transport_types()
+        connection = https_connection(host="objects.example.test", port=443)
+        with patch(
+            "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+            return_value=_Socket(),
+        ) as connector:
+            connection._new_conn()
+
+        self.assertEqual("objects.example.test", connection.host)
+        self.assertEqual("objects.example.test", connection._dns_host)
+        self.assertEqual("objects.example.test", connector.call_args.args[0])
+
+    def test_mixed_answer_and_peer_change_fail_closed(self) -> None:
+        _, http_connection, _ = _botocore_transport_types()
+        connection = http_connection(host="objects.example.test", port=80)
+        records = [
+            [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80))],
+            [
+                (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 80)),
+                (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 80)),
+            ],
+        ]
+        with patch.dict(
+            "os.environ",
+            {"APP_ENV": "production", "GOVOPLAN_CONNECTOR_ALLOW_PRIVATE_NETWORKS": "false"},
+        ), patch(
+            "govoplan_core.security.outbound_http.socket.getaddrinfo",
+            side_effect=records,
+        ), patch(
+            "govoplan_core.security.outbound_http.socket.socket",
+            return_value=_Socket(),
+        ):
+            connection._new_conn()
+            with self.assertRaisesRegex(Exception, "non-public network"):
+                connection._new_conn()
+
+
+class _FakeTcp:
+    def __init__(self, server: str, port: int, timeout: float | None = None) -> None:
+        self.server = server
+        self.port = port
+        self.timeout = timeout
+        self.connected = False
+        self._sock = None
+        self._sock_lock = threading.Lock()
+
+
+class _FakeConnection:
+    def __init__(self, _guid: object, server: str, port: int, *, require_signing: bool) -> None:
+        self.server = server
+        self.port = port
+        self.require_signing = require_signing
+        self.transport = _FakeTcp(server, port)
+        self.session_table: dict[str, object] = {}
+
+    def connect(self, *, timeout: float) -> None:
+        self.transport.timeout = timeout
+        self.transport.connect()
+
+
+class _FakeSession:
+    def __init__(
+        self,
+        connection: _FakeConnection,
+        *,
+        username: str | None,
+        password: str | None,
+        require_encryption: bool,
+        auth_protocol: str,
+    ) -> None:
+        self.connection = connection
+        self.username = username
+        self.password = password
+        self.encrypt_data = require_encryption
+        self.auth_protocol = auth_protocol
+        self.encrypt = require_encryption
+
+    def connect(self) -> None:
+        self.connection.session_table[self.username or "anonymous"] = self
+
+
+class SmbPeerPinningTests(unittest.TestCase):
+    def setUp(self) -> None:
+        pinned_smb_connection_cache.cache_clear()
+
+    def _modules(self) -> tuple[types.ModuleType, dict[str, types.ModuleType]]:
+        smbclient = types.ModuleType("smbclient")
+        pool = types.ModuleType("smbclient._pool")
+
+        def original_register_session(
+            server: str,
+            username: str | None = None,
+            password: str | None = None,
+            port: int = 445,
+            encrypt: bool | None = None,
+            connection_timeout: float = 60,
+            connection_cache: dict[str, object] | None = None,
+            auth_protocol: str = "negotiate",
+            require_signing: bool = True,
+        ) -> None:
+            del server, username, password, port, encrypt, connection_timeout
+            del connection_cache, auth_protocol, require_signing
+
+        pool.register_session = original_register_session
+        pool.ClientConfig = lambda: types.SimpleNamespace(client_guid="guid")
+        connection = types.ModuleType("smbprotocol.connection")
+        connection.Connection = _FakeConnection
+        connection.Tcp = _FakeTcp
+        session = types.ModuleType("smbprotocol.session")
+        session.Session = _FakeSession
+        transport = types.ModuleType("smbprotocol.transport")
+        transport.Tcp = _FakeTcp
+        return smbclient, {
+            "smbclient._pool": pool,
+            "smbprotocol.connection": connection,
+            "smbprotocol.session": session,
+            "smbprotocol.transport": transport,
+        }
+
+    def test_initial_reconnect_and_referral_hosts_are_each_pinned(self) -> None:
+        smbclient, modules = self._modules()
+        sockets = [_Socket(), _Socket(), _Socket()]
+        with patch.dict(sys.modules, modules), patch(
+            "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+            side_effect=sockets,
+        ) as connector:
+            install_pinned_smb_transport(smbclient)
+            register = modules["smbclient._pool"].register_session
+            cache: dict[str, object] = {}
+            first = register("files.example.test", connection_cache=cache)
+            first.connection.transport.connected = False
+            register("files.example.test", connection_cache=cache)
+            register("dfs-target.example.test", connection_cache=cache)
+
+        self.assertEqual(
+            ["files.example.test", "files.example.test", "dfs-target.example.test"],
+            [call.args[0] for call in connector.call_args_list],
+        )
+
+    def test_referral_to_disallowed_peer_fails_before_session_creation(self) -> None:
+        smbclient, modules = self._modules()
+        with patch.dict(sys.modules, modules), patch(
+            "govoplan_files.backend.storage.sdk_peer_pinning.create_outbound_connection",
+            side_effect=OutboundHttpBlocked("non-public network"),
+        ):
+            install_pinned_smb_transport(smbclient)
+            register = modules["smbclient._pool"].register_session
+            with self.assertRaisesRegex(ValueError, "non-public network"):
+                register("private-referral.example.test", connection_cache={})
+
+    def test_transport_tampering_after_installation_fails_closed(self) -> None:
+        smbclient, modules = self._modules()
+        with patch.dict(sys.modules, modules):
+            install_pinned_smb_transport(smbclient)
+            modules["smbprotocol.connection"].Tcp = _FakeTcp
+            with self.assertRaisesRegex(SdkPeerPinningError, "changed after peer pinning"):
+                install_pinned_smb_transport(smbclient)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_search_source.py

@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from govoplan_access.backend.db.models import Account, Group, User
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import PrincipalRef
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
+from govoplan_core.core.events import EventObjectRef, EventTenantRef, PlatformEvent
+from govoplan_core.core.search import (
+    SearchAuthorizationRequest,
+    SearchBackfillRequest,
+    SearchResourceReference,
+)
+from govoplan_core.db.base import Base
+from govoplan_files.backend.db.models import FileAsset, FileFolder, FileShare
+from govoplan_files.backend.search_source import (
+    FilesSearchSource,
+    PROVIDER_ID,
+)
+
+
+class FilesSearchSourceTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite://")
+        Base.metadata.create_all(
+            self.engine,
+            tables=(
+                Account.__table__,
+                User.__table__,
+                Group.__table__,
+                FileAsset.__table__,
+                FileFolder.__table__,
+                FileShare.__table__,
+                ChangeSequenceEntry.__table__,
+            ),
+        )
+        self.session = Session(self.engine)
+        self.session.add_all(
+            (
+                Account(
+                    id="account-1",
+                    email="one@example.test",
+                    normalized_email="one@example.test",
+                ),
+                User(
+                    id="user-1",
+                    tenant_id="tenant-1",
+                    account_id="account-1",
+                    email="one@example.test",
+                ),
+                FileAsset(
+                    id="file-1",
+                    tenant_id="tenant-1",
+                    owner_type="user",
+                    owner_user_id="user-1",
+                    display_path="records/permit.pdf",
+                    filename="permit.pdf",
+                    description="Monthly permit evidence",
+                ),
+                FileAsset(
+                    id="file-other",
+                    tenant_id="tenant-2",
+                    owner_type="user",
+                    display_path="other.pdf",
+                    filename="other.pdf",
+                ),
+            )
+        )
+        self.session.commit()
+        self.source = FilesSearchSource()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        self.engine.dispose()
+
+    def test_backfill_and_authorization_are_tenant_bounded(self) -> None:
+        page = self.source.backfill(
+            self.session,
+            request=SearchBackfillRequest(
+                tenant_id="tenant-1",
+                provider_id=PROVIDER_ID,
+                resource_type="file",
+                rebuild_id="rebuild-1",
+            ),
+        )
+        self.assertEqual(("file-1",), tuple(doc.resource_id for doc in page.documents))
+        reference = SearchResourceReference(
+            tenant_id="tenant-1",
+            module_id="files",
+            resource_type="file",
+            resource_id="file-1",
+        )
+        request = SearchAuthorizationRequest(reference=reference, source_revision="1")
+        self.assertTrue(
+            self.source.authorize(
+                self.session,
+                _principal({"files:file:read"}),
+                requests=(request,),
+            )[reference.key]
+        )
+        self.assertFalse(
+            self.source.authorize(
+                self.session,
+                _principal(set()),
+                requests=(request,),
+            )[reference.key]
+        )
+
+    def test_committed_file_event_produces_authoritative_upsert(self) -> None:
+        event = PlatformEvent(
+            type="files.file.updated",
+            module_id="files",
+            tenant=EventTenantRef(id="tenant-1"),
+            resource=EventObjectRef(type="file", id="file-1"),
+        )
+        changes = self.source.index_changes_for_event(
+            self.session,
+            event=event,
+            delivery_key="delivery-1",
+        )
+        self.assertEqual(1, len(changes))
+        self.assertEqual("upsert", changes[0].kind)
+        self.assertEqual(event.event_id, changes[0].cursor)
+
+
+def _principal(scopes: set[str]) -> ApiPrincipal:
+    return ApiPrincipal(
+        principal=PrincipalRef(
+            account_id="account-1",
+            membership_id="user-1",
+            tenant_id="tenant-1",
+            scopes=frozenset(scopes),
+        ),
+        account=SimpleNamespace(id="account-1"),
+        user=SimpleNamespace(id="user-1"),
+    )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_share_lifecycle.py

@@ -5,6 +5,7 @@ from datetime import timedelta
 from unittest.mock import patch
 
 from sqlalchemy import create_engine
+from sqlalchemy.dialects import postgresql
 from sqlalchemy.orm import sessionmaker
 
 from govoplan_access.backend.db.models import Account, Group, User
@@ -13,6 +14,7 @@ from govoplan_core.db.base import Base
 from govoplan_files.backend.db.models import FileAsset, FileShare
 from govoplan_files.backend.storage.common import FileStorageError, utcnow
 from govoplan_files.backend.storage.files import (
+    _asset_visibility_query_for_user,
     get_asset_for_user,
     list_file_shares,
     revoke_file_share,
@@ -214,6 +216,22 @@ class FileShareLifecycleTests(unittest.TestCase):
                 expires_at=utcnow() - timedelta(seconds=1),
             )
 
+    @patch(
+        "govoplan_files.backend.storage.files.user_group_ids",
+        return_value=[],
+    )
+    def test_visibility_query_is_postgresql_json_safe(self, _groups) -> None:
+        query = _asset_visibility_query_for_user(
+            self.session,
+            tenant_id=TENANT_ID,
+            user_id=RECIPIENT_ID,
+        )
+
+        compiled = str(query.statement.compile(dialect=postgresql.dialect()))
+
+        self.assertNotIn("SELECT DISTINCT", compiled.upper())
+        self.assertIn("EXISTS", compiled.upper())
+
 
 if __name__ == "__main__":
     unittest.main()

webui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@govoplan/files-webui",
-  "version": "0.1.9",
+  "version": "0.1.17",
   "private": true,
   "type": "module",
   "main": "src/index.ts",
@@ -26,7 +26,7 @@
     "react-dom": ">=19.2.7 <20",
     "react-router": ">=8.3.0 <9",
     "lucide-react": "^1.23.0",
-    "@govoplan/core-webui": "^0.1.9"
+    "@govoplan/core-webui": "^0.1.17"
   },
   "peerDependenciesMeta": {
     "@govoplan/core-webui": {

webui/scripts/test-interface-pattern-language.mjs

@@ -7,6 +7,8 @@ function read(relativePath) {
 }
 
 const connector = read("../src/features/files/FileConnectorSettingsPanel.tsx");
+const integrity = read("../src/features/files/FileIntegrityPanel.tsx");
+const filesApi = read("../src/api/files.ts");
 const filesPage = read("../src/features/files/FilesPage.tsx");
 const moduleSource = read("../src/module.ts");
 const styles = read("../src/styles/file-manager.css");
@@ -23,6 +25,14 @@ assert.match(connector, /<AdvancedOptionsPanel title="Connection metadata JSON"/
 assert.match(connector, /<AdvancedOptionsPanel title="Credential metadata JSON"/);
 assert.doesNotMatch(connector, /window\.(?:alert|confirm)\(/);
 
+assert.match(integrity, /File integrity/);
+assert.match(filesApi, /expected_revision/);
+assert.match(integrity, /cleanupFileIntegrityFinding\(settings, finding, true\)/);
+assert.match(integrity, /<ConfirmDialog[\s\S]*Delete unreferenced storage object/);
+assert.match(integrity, /files\.reference\.integrity-recovery-and-fail-closed-transports/);
+assert.match(moduleSource, /files\.admin\.tenant-integrity/);
+assert.doesNotMatch(integrity, /window\.(?:alert|confirm)\(/);
+
 assert.match(filesPage, /DocumentationHelpLink/);
 assert.match(filesPage, /topicId: "files\.workflow\.organize-managed-files"/);
 assert.match(filesPage, /disabledReason=\{uploadBlocker\}/);
@@ -31,7 +41,7 @@ assert.match(filesPage, /<ConfirmDialog[\s\S]*tone="danger"/);
 assert.match(filesPage, /className="workspace-data-page module-entry-page file-manager-page file-manager-fullscreen files-page"/);
 assert.doesNotMatch(filesPage, /window\.(?:alert|confirm)\(/);
 
-assert.doesNotMatch(`${connector}\n${filesPage}\n${moduleSource}`, /@govoplan\/(?:campaign|mail|docs)-webui|govoplan_(?:campaign|mail|docs)/);
+assert.doesNotMatch(`${connector}\n${integrity}\n${filesPage}\n${moduleSource}`, /@govoplan\/(?:campaign|mail|docs)-webui|govoplan_(?:campaign|mail|docs)/);
 assert.match(moduleSource, /"files\.connectors"/);
 assert.match(moduleSource, /"files\.fileExplorer"/);
 assert.match(styles, /@media \(max-width: 1050px\)[\s\S]*\.files-page \.file-manager-shell[\s\S]*grid-template-columns: 1fr/);

webui/src/api/files.ts

@@ -207,6 +207,57 @@ export type FileConnectorCredentialUpdatePayload = Partial<Omit<FileConnectorCre
   clear_token?: boolean;
 };
 
+export type FileIntegrityScan = {
+  id: string;
+  tenant_id: string;
+  storage_backend: string;
+  storage_prefix: string;
+  status: string;
+  revision: number;
+  phase: string;
+  verify_checksums: boolean;
+  batch_size: number;
+  scanned_blob_count: number;
+  verified_blob_count: number;
+  quarantined_blob_count: number;
+  scanned_object_count: number;
+  orphan_object_count: number;
+  created_by_user_id?: string | null;
+  started_at?: string | null;
+  completed_at?: string | null;
+  last_error?: string | null;
+  created_at: string;
+  updated_at: string;
+};
+
+export type FileIntegrityFinding = {
+  id: string;
+  scan_id: string;
+  tenant_id: string;
+  kind: string;
+  state: string;
+  revision: number;
+  blob_id?: string | null;
+  storage_key: string;
+  expected_size_bytes?: number | null;
+  observed_size_bytes?: number | null;
+  expected_checksum_sha256?: string | null;
+  observed_checksum_sha256?: string | null;
+  resolved_at?: string | null;
+  resolved_by_user_id?: string | null;
+  created_at: string;
+  updated_at: string;
+};
+
+export type FileIntegrityActionResult = {
+  action: string;
+  changed: boolean;
+  dry_run: boolean;
+  finding: FileIntegrityFinding;
+  inspection_kind?: string | null;
+  inspection_valid?: boolean | null;
+};
+
 export type FileConnectorBrowseItem = {
   kind: "library" | "folder" | "file";
   name: string;
@@ -1039,6 +1090,91 @@ payload: {
   return apiFetch<TransferResponse>(settings, "/api/v1/files/transfer", { method: "POST", body: JSON.stringify(payload) });
 }
 
+export async function listFileIntegrityScans(
+  settings: ApiSettings,
+  limit = 100
+): Promise<FileIntegrityScan[]> {
+  const response = await apiFetch<{ scans: FileIntegrityScan[] }>(
+    settings,
+    `/api/v1/files/integrity/scans?limit=${Math.max(1, Math.min(limit, 200))}`
+  );
+  return response.scans;
+}
+
+export function createFileIntegrityScan(
+  settings: ApiSettings,
+  payload: { verify_checksums: boolean; batch_size: number }
+): Promise<FileIntegrityScan> {
+  return apiFetch<FileIntegrityScan>(settings, "/api/v1/files/integrity/scans", {
+    method: "POST",
+    body: JSON.stringify(payload)
+  });
+}
+
+export function runFileIntegrityScanBatch(
+  settings: ApiSettings,
+  scan: Pick<FileIntegrityScan, "id" | "revision">
+): Promise<FileIntegrityScan> {
+  return apiFetch<FileIntegrityScan>(
+    settings,
+    `/api/v1/files/integrity/scans/${encodeURIComponent(scan.id)}/run`,
+    {
+      method: "POST",
+      body: JSON.stringify({ expected_revision: scan.revision })
+    }
+  );
+}
+
+export async function listFileIntegrityFindings(
+  settings: ApiSettings,
+  scanId: string,
+  state?: string
+): Promise<FileIntegrityFinding[]> {
+  const search = new URLSearchParams({ limit: "1000" });
+  if (state) search.set("state", state);
+  const response = await apiFetch<{ findings: FileIntegrityFinding[] }>(
+    settings,
+    `/api/v1/files/integrity/scans/${encodeURIComponent(scanId)}/findings?${search}`
+  );
+  return response.findings;
+}
+
+export function recheckFileIntegrityFinding(
+  settings: ApiSettings,
+  finding: Pick<FileIntegrityFinding, "id" | "revision">,
+  dryRun = false
+): Promise<FileIntegrityActionResult> {
+  return apiFetch<FileIntegrityActionResult>(
+    settings,
+    `/api/v1/files/integrity/findings/${encodeURIComponent(finding.id)}/recheck`,
+    {
+      method: "POST",
+      body: JSON.stringify({
+        dry_run: dryRun,
+        expected_revision: finding.revision
+      })
+    }
+  );
+}
+
+export function cleanupFileIntegrityFinding(
+  settings: ApiSettings,
+  finding: Pick<FileIntegrityFinding, "id" | "revision">,
+  dryRun: boolean
+): Promise<FileIntegrityActionResult> {
+  return apiFetch<FileIntegrityActionResult>(
+    settings,
+    `/api/v1/files/integrity/findings/${encodeURIComponent(finding.id)}/cleanup`,
+    {
+      method: "POST",
+      body: JSON.stringify({
+        dry_run: dryRun,
+        expected_revision: finding.revision
+      })
+    }
+  );
+}
+
 export async function downloadFile(settings: ApiSettings, file: ManagedFile): Promise<void> {
   const response = await fetch(apiUrl(settings, `/api/v1/files/${file.id}/download`), { headers: authHeaders(settings), credentials: "include" });
   if (!response.ok) throw new Error(`${response.status} ${response.statusText}: ${await response.text()}`);

webui/src/features/files/FileIntegrityPanel.tsx

@@ -0,0 +1,527 @@
+import { useEffect, useMemo, useState } from "react";
+import {
+  AdminPageLayout,
+  adminErrorMessage,
+  Button,
+  Card,
+  ConfirmDialog,
+  DataGrid,
+  Dialog,
+  DismissibleAlert,
+  DocumentationHelpLink,
+  FormField,
+  MetricCard,
+  StatusBadge,
+  TableActionGroup,
+  ToggleSwitch,
+  type ApiSettings,
+  type DataGridColumn
+} from "@govoplan/core-webui";
+import {
+  CheckCircle2,
+  Eye,
+  Play,
+  Plus,
+  RefreshCw,
+  RotateCw,
+  Trash2
+} from "lucide-react";
+import {
+  cleanupFileIntegrityFinding,
+  createFileIntegrityScan,
+  listFileIntegrityFindings,
+  listFileIntegrityScans,
+  recheckFileIntegrityFinding,
+  runFileIntegrityScanBatch,
+  type FileIntegrityActionResult,
+  type FileIntegrityFinding,
+  type FileIntegrityScan
+} from "../../api/files";
+
+type Props = {
+  settings: ApiSettings;
+  canWrite: boolean;
+};
+
+const DOCUMENTATION = {
+  topicId: "files.reference.integrity-recovery-and-fail-closed-transports",
+  documentationType: "admin" as const
+};
+
+export default function FileIntegrityPanel({ settings, canWrite }: Props) {
+  const [scans, setScans] = useState<FileIntegrityScan[]>([]);
+  const [selectedScanId, setSelectedScanId] = useState("");
+  const [findings, setFindings] = useState<FileIntegrityFinding[]>([]);
+  const [loading, setLoading] = useState(true);
+  const [busy, setBusy] = useState(false);
+  const [error, setError] = useState("");
+  const [success, setSuccess] = useState("");
+  const [createOpen, setCreateOpen] = useState(false);
+  const [verifyChecksums, setVerifyChecksums] = useState(true);
+  const [batchSize, setBatchSize] = useState(100);
+  const [cleanupPreview, setCleanupPreview] = useState<FileIntegrityActionResult | null>(null);
+
+  const selectedScan = scans.find((scan) => scan.id === selectedScanId) ?? null;
+
+  useEffect(() => {
+    void loadScans();
+  }, [settings.accessToken, settings.apiBaseUrl, settings.apiKey]);
+
+  useEffect(() => {
+    if (!selectedScanId) {
+      setFindings([]);
+      return;
+    }
+    void loadFindings(selectedScanId);
+  }, [selectedScanId, settings.accessToken, settings.apiBaseUrl, settings.apiKey]);
+
+  async function loadScans() {
+    setLoading(true);
+    setError("");
+    try {
+      const loaded = await listFileIntegrityScans(settings);
+      setScans(loaded);
+      setSelectedScanId((current) => (
+        loaded.some((scan) => scan.id === current) ? current : loaded[0]?.id ?? ""
+      ));
+    } catch (err) {
+      setError(adminErrorMessage(err));
+    } finally {
+      setLoading(false);
+    }
+  }
+
+  async function loadFindings(scanId: string) {
+    setError("");
+    try {
+      setFindings(await listFileIntegrityFindings(settings, scanId));
+    } catch (err) {
+      setError(adminErrorMessage(err));
+    }
+  }
+
+  function replaceScan(next: FileIntegrityScan) {
+    setScans((current) => [
+      next,
+      ...current.filter((scan) => scan.id !== next.id)
+    ].sort((left, right) => right.created_at.localeCompare(left.created_at)));
+  }
+
+  function replaceFinding(next: FileIntegrityFinding) {
+    setFindings((current) => current.map((finding) => (
+      finding.id === next.id ? next : finding
+    )));
+  }
+
+  async function createScan() {
+    setBusy(true);
+    setError("");
+    setSuccess("");
+    try {
+      const scan = await createFileIntegrityScan(settings, {
+        verify_checksums: verifyChecksums,
+        batch_size: batchSize
+      });
+      replaceScan(scan);
+      setSelectedScanId(scan.id);
+      setCreateOpen(false);
+      setSuccess("Integrity scan created. Run its first bounded batch when ready.");
+    } catch (err) {
+      setError(adminErrorMessage(err));
+    } finally {
+      setBusy(false);
+    }
+  }
+
+  async function runBatch(scan: FileIntegrityScan) {
+    setBusy(true);
+    setError("");
+    setSuccess("");
+    try {
+      const next = await runFileIntegrityScanBatch(settings, scan);
+      replaceScan(next);
+      setSelectedScanId(next.id);
+      await loadFindings(next.id);
+      setSuccess(
+        next.status === "completed"
+          ? "Integrity scan completed. Review and resolve every finding."
+          : `Completed one ${next.batch_size}-item batch; the scan can be resumed.`
+      );
+    } catch (err) {
+      setError(adminErrorMessage(err));
+      await loadScans();
+    } finally {
+      setBusy(false);
+    }
+  }
+
+  async function recheck(finding: FileIntegrityFinding) {
+    setBusy(true);
+    setError("");
+    setSuccess("");
+    try {
+      const result = await recheckFileIntegrityFinding(settings, finding);
+      replaceFinding(result.finding);
+      setSuccess(
+        result.inspection_valid
+          ? "The stored object now matches its recorded integrity evidence."
+          : "The object still fails integrity verification and remains quarantined."
+      );
+    } catch (err) {
+      setError(adminErrorMessage(err));
+      if (selectedScanId) await loadFindings(selectedScanId);
+    } finally {
+      setBusy(false);
+    }
+  }
+
+  async function previewCleanup(finding: FileIntegrityFinding) {
+    setBusy(true);
+    setError("");
+    setSuccess("");
+    try {
+      const preview = await cleanupFileIntegrityFinding(settings, finding, true);
+      setCleanupPreview(preview);
+      if (preview.action !== "would_delete") {
+        replaceFinding(preview.finding);
+        setSuccess(cleanupActionMessage(preview.action));
+      }
+    } catch (err) {
+      setError(adminErrorMessage(err));
+      if (selectedScanId) await loadFindings(selectedScanId);
+    } finally {
+      setBusy(false);
+    }
+  }
+
+  async function confirmCleanup() {
+    if (!cleanupPreview) return;
+    setBusy(true);
+    setError("");
+    setSuccess("");
+    try {
+      const result = await cleanupFileIntegrityFinding(
+        settings,
+        cleanupPreview.finding,
+        false
+      );
+      replaceFinding(result.finding);
+      setCleanupPreview(null);
+      setSuccess(cleanupActionMessage(result.action));
+    } catch (err) {
+      setCleanupPreview(null);
+      setError(adminErrorMessage(err));
+      if (selectedScanId) await loadFindings(selectedScanId);
+    } finally {
+      setBusy(false);
+    }
+  }
+
+  const scanColumns = useMemo<DataGridColumn<FileIntegrityScan>[]>(() => [
+    {
+      id: "created",
+      header: "Created",
+      width: 180,
+      minWidth: 150,
+      sortable: true,
+      value: (scan) => scan.created_at,
+      render: (scan) => formatDateTime(scan.created_at)
+    },
+    {
+      id: "status",
+      header: "Status",
+      width: 130,
+      minWidth: 110,
+      sortable: true,
+      filterable: true,
+      filterType: "list",
+      value: (scan) => scan.status,
+      render: (scan) => <StatusBadge status={statusTone(scan.status)} label={scan.status} />
+    },
+    {
+      id: "phase",
+      header: "Phase",
+      width: 110,
+      minWidth: 90,
+      value: (scan) => scan.phase
+    },
+    {
+      id: "progress",
+      header: "Progress",
+      minWidth: 250,
+      resizable: true,
+      value: (scan) => `${scan.scanned_blob_count} blobs, ${scan.scanned_object_count} objects`,
+      render: (scan) => (
+        <span>
+          {scan.scanned_blob_count} blobs ({scan.verified_blob_count} verified, {scan.quarantined_blob_count} quarantined), {" "}
+          {scan.scanned_object_count} objects ({scan.orphan_object_count} orphaned)
+        </span>
+      )
+    },
+    {
+      id: "backend",
+      header: "Storage",
+      width: 120,
+      minWidth: 100,
+      value: (scan) => scan.storage_backend
+    },
+    {
+      id: "actions",
+      header: "Actions",
+      width: 105,
+      minWidth: 105,
+      sticky: "end",
+      align: "right",
+      render: (scan) => (
+        <TableActionGroup
+          minimumSlots={2}
+          actions={[
+            {
+              id: "inspect",
+              label: "Inspect findings",
+              icon: <Eye size={16} />,
+              onClick: () => setSelectedScanId(scan.id)
+            },
+            {
+              id: "run",
+              label: scan.status === "failed" ? "Resume next batch" : "Run next batch",
+              icon: <Play size={16} />,
+              onClick: () => void runBatch(scan),
+              disabled: busy || ["completed", "cancelled"].includes(scan.status),
+              disabledReason: ["completed", "cancelled"].includes(scan.status)
+                ? "This scan is complete."
+                : undefined
+            }
+          ]}
+        />
+      )
+    }
+  ], [busy]);
+
+  const findingColumns = useMemo<DataGridColumn<FileIntegrityFinding>[]>(() => [
+    {
+      id: "kind",
+      header: "Finding",
+      width: 180,
+      minWidth: 145,
+      sortable: true,
+      filterable: true,
+      value: (finding) => finding.kind.replaceAll("_", " ")
+    },
+    {
+      id: "state",
+      header: "State",
+      width: 115,
+      minWidth: 95,
+      sortable: true,
+      filterable: true,
+      value: (finding) => finding.state,
+      render: (finding) => <StatusBadge status={statusTone(finding.state)} label={finding.state} />
+    },
+    {
+      id: "object",
+      header: "Affected object",
+      minWidth: 260,
+      resizable: true,
+      filterable: true,
+      value: (finding) => finding.storage_key,
+      render: (finding) => (
+        <span title={finding.storage_key}>
+          {finding.blob_id ? `Blob ${finding.blob_id}` : finding.storage_key}
+        </span>
+      )
+    },
+    {
+      id: "evidence",
+      header: "Expected / observed",
+      minWidth: 230,
+      resizable: true,
+      value: (finding) => evidenceLabel(finding)
+    },
+    {
+      id: "updated",
+      header: "Updated",
+      width: 175,
+      minWidth: 145,
+      sortable: true,
+      value: (finding) => finding.updated_at,
+      render: (finding) => formatDateTime(finding.updated_at)
+    },
+    {
+      id: "actions",
+      header: "Actions",
+      width: 70,
+      minWidth: 70,
+      sticky: "end",
+      align: "right",
+      render: (finding) => (
+        <TableActionGroup
+          minimumSlots={1}
+          actions={finding.kind === "orphan_object"
+            ? [{
+                id: "cleanup",
+                label: "Preview safe cleanup",
+                icon: <Trash2 size={16} />,
+                variant: "danger",
+                onClick: () => void previewCleanup(finding),
+                disabled: busy || finding.state === "deleted",
+                disabledReason: finding.state === "deleted" ? "The object is already absent." : undefined
+              }]
+            : [{
+                id: "recheck",
+                label: "Recheck stored object",
+                icon: <RotateCw size={16} />,
+                onClick: () => void recheck(finding),
+                disabled: busy
+              }]}
+        />
+      )
+    }
+  ], [busy]);
+
+  return (
+    <>
+      <AdminPageLayout
+        title="File integrity"
+        description="Run bounded storage reconciliation and resolve quarantined or unreferenced objects without acting on stale operator state."
+        loading={loading}
+        error={error}
+        success={success}
+        actions={(
+          <>
+            <Button
+              title="Reload scans and findings"
+              aria-label="Reload scans and findings"
+              onClick={() => void loadScans()}
+              disabled={loading || busy}
+            >
+              <RefreshCw size={16} />
+            </Button>
+            <Button variant="primary" onClick={() => setCreateOpen(true)} disabled={!canWrite || busy}>
+              <Plus size={16} /> New scan
+            </Button>
+            <DocumentationHelpLink reference={DOCUMENTATION} label="Open Files integrity documentation" />
+          </>
+        )}
+      >
+        <DismissibleAlert tone="info" dismissible={false} compact>
+          Cleanup is available only for objects with no managed database reference and always starts with a dry-run preview. Files does not yet implement legal-hold or hard-purge policy; such objects must remain outside cleanup until those controls exist.
+        </DismissibleAlert>
+
+        <Card title="Integrity scans">
+          <div className="admin-table-surface">
+            <DataGrid
+              id="files-integrity-scans-v1"
+              rows={scans}
+              columns={scanColumns}
+              getRowKey={(scan) => scan.id}
+              initialFit="container"
+              rowClassName={(scan) => scan.id === selectedScanId ? "is-selected" : undefined}
+              emptyText="No integrity scans have been created."
+            />
+          </div>
+        </Card>
+
+        {selectedScan && (
+          <>
+            <div className="metric-grid">
+              <MetricCard label="Verified blobs" value={selectedScan.verified_blob_count} tone="good" />
+              <MetricCard label="Quarantined blobs" value={selectedScan.quarantined_blob_count} tone={selectedScan.quarantined_blob_count ? "danger" : "neutral"} />
+              <MetricCard label="Orphan objects" value={selectedScan.orphan_object_count} tone={selectedScan.orphan_object_count ? "warning" : "neutral"} />
+              <MetricCard label="Open findings" value={findings.filter((finding) => finding.state === "open").length} tone={findings.some((finding) => finding.state === "open") ? "warning" : "good"} />
+            </div>
+            {selectedScan.last_error && (
+              <DismissibleAlert tone="warning" dismissible={false} compact>
+                The last batch failed with {selectedScan.last_error}. Verify storage availability, then resume from the recorded cursor.
+              </DismissibleAlert>
+            )}
+            <Card title="Findings">
+              <div className="admin-table-surface">
+                <DataGrid
+                  id={`files-integrity-findings-${selectedScan.id}`}
+                  rows={findings}
+                  columns={findingColumns}
+                  getRowKey={(finding) => finding.id}
+                  initialFit="container"
+                  emptyText={selectedScan.status === "completed" ? "No findings were recorded." : "No findings in completed batches yet."}
+                />
+              </div>
+            </Card>
+          </>
+        )}
+      </AdminPageLayout>
+
+      <Dialog
+        open={createOpen}
+        title="Create integrity scan"
+        onClose={() => setCreateOpen(false)}
+        closeDisabled={busy}
+        footer={(
+          <>
+            <Button onClick={() => setCreateOpen(false)} disabled={busy}>Cancel</Button>
+            <Button variant="primary" onClick={() => void createScan()} disabled={busy}>
+              <CheckCircle2 size={16} /> {busy ? "Creating..." : "Create scan"}
+            </Button>
+          </>
+        )}
+      >
+        <div className="settings-list">
+          <ToggleSwitch
+            label="Verify SHA-256 checksums"
+            checked={verifyChecksums}
+            onChange={setVerifyChecksums}
+            disabled={busy}
+            help="Checksum verification reads every selected object; disabling it verifies existence and size only."
+          />
+          <FormField label="Items per batch" documentation={DOCUMENTATION}>
+            <input
+              type="number"
+              min={1}
+              max={1000}
+              value={batchSize}
+              onChange={(event) => setBatchSize(Math.max(1, Math.min(1000, Number(event.target.value) || 1)))}
+              disabled={busy}
+            />
+          </FormField>
+        </div>
+      </Dialog>
+
+      <ConfirmDialog
+        open={cleanupPreview?.action === "would_delete"}
+        title="Delete unreferenced storage object?"
+        message={cleanupPreview ? `The dry run confirmed that ${cleanupPreview.finding.storage_key} has no managed Files reference. Deletion cannot be undone from GovOPlaN.` : ""}
+        confirmLabel="Delete object"
+        tone="danger"
+        busy={busy}
+        onConfirm={() => void confirmCleanup()}
+        onCancel={() => setCleanupPreview(null)}
+      />
+    </>
+  );
+}
+
+function statusTone(value: string): string {
+  if (["completed", "resolved", "verified"].includes(value)) return "success";
+  if (["failed", "deleted", "missing", "checksum_mismatch", "size_mismatch"].includes(value)) return "danger";
+  if (["running", "pending", "open"].includes(value)) return "warning";
+  return "neutral";
+}
+
+function evidenceLabel(finding: FileIntegrityFinding): string {
+  const expected = finding.expected_size_bytes == null ? "-" : `${finding.expected_size_bytes} B`;
+  const observed = finding.observed_size_bytes == null ? "-" : `${finding.observed_size_bytes} B`;
+  return `${expected} / ${observed}`;
+}
+
+function cleanupActionMessage(action: string): string {
+  if (action === "retained_referenced") return "Cleanup was blocked because the object has a managed database reference.";
+  if (action === "already_deleted" || action === "already_absent") return "The object was already absent; the finding is reconciled.";
+  if (action === "deleted") return "The unreferenced storage object was deleted and recovery evidence was recorded.";
+  return `Integrity cleanup result: ${action.replaceAll("_", " ")}.`;
+}
+
+function formatDateTime(value: string | null | undefined): string {
+  if (!value) return "-";
+  const parsed = new Date(value);
+  return Number.isNaN(parsed.getTime()) ? value : parsed.toLocaleString();
+}

webui/src/index.ts

@@ -1,6 +1,7 @@
 export { default } from "./module";
 export * from "./module";
 export { default as FilesPage } from "./features/files/FilesPage";
+export { default as FileIntegrityPanel } from "./features/files/FileIntegrityPanel";
 export * from "./api/files";
 export type { PlatformWebModule, PlatformNavItem, PlatformRouteContribution, PlatformRouteContext } from "@govoplan/core-webui";
 export { FolderTree } from "./features/files/components/FileManagerComponents";

webui/src/module.ts

@@ -16,6 +16,7 @@ import { generatedTranslations } from "./i18n/generatedTranslations";
 import "./styles/file-manager.css";
 
 const FilesPage = lazy(() => import("./features/files/FilesPage"));
+const FileIntegrityPanel = lazy(() => import("./features/files/FileIntegrityPanel"));
 
 const fileRead = ["files:file:read"];
 const translations = {
@@ -97,6 +98,20 @@ const fileConnectorAdminSections: AdminSectionsUiCapability = {
       scopeType: "tenant",
       canWrite: hasScope(auth, "admin:settings:write") || hasScope(auth, "files:file:admin")
     })
+  },
+  {
+    id: "tenant-file-integrity",
+    moduleId: "files",
+    kind: "operations",
+    surfaceId: "files.admin.tenant-integrity",
+    label: "File integrity",
+    group: "TENANT",
+    order: 66,
+    allOf: ["files:file:admin"],
+    render: ({ settings, auth }) => createElement(FileIntegrityPanel, {
+      settings,
+      canWrite: hasScope(auth, "files:file:admin")
+    })
   }]
 
 };
@@ -110,6 +125,7 @@ export const filesModule: PlatformWebModule = {
   viewSurfaces: [
     { id: "files.admin.system-connectors", moduleId: "files", kind: "section", label: "System file connections", order: 75 },
     { id: "files.admin.tenant-connectors", moduleId: "files", kind: "section", label: "Tenant file connections", order: 65 },
+    { id: "files.admin.tenant-integrity", moduleId: "files", kind: "section", label: "File integrity", order: 66 },
     { id: "files.admin.group-connectors", moduleId: "files", kind: "section", label: "Group file connections", order: 65 },
     { id: "files.admin.user-connectors", moduleId: "files", kind: "section", label: "User file connections", order: 65 },
     { id: "files.settings.connectors", moduleId: "files", kind: "section", label: "Personal file connections", order: 20 },