From 27f3603d8fd14c94692bbb4b33c26f458245e68d Mon Sep 17 00:00:00 2001 From: Anatolii Hrynchuk Date: Sat, 3 Oct 2026 00:56:28 +0300 Subject: [PATCH] chore: preserve current workspace changes Save pending configuration, deployment, and documentation work. --- .dockerignore | 19 +- README.md | 241 +++--- .../kubernetes/shared-notification/.gitignore | 3 + .../kubernetes/shared-notification/README.md | 331 +++++++++ .../artifacts.example.json | 14 + .../shared-notification/config.example.json | 19 + .../shared-notification/notification.py | 695 ++++++++++++++++++ .../shared-notification/prepare-seq-key.py | 276 +++++++ .../shared-notification/prepare-storage.py | 88 +++ .../tests/test_notification.py | 446 +++++++++++ .../shared-notification/tests/test_seq_key.py | 317 ++++++++ 11 files changed, 2320 insertions(+), 129 deletions(-) create mode 100644 docker/kubernetes/shared-notification/.gitignore create mode 100644 docker/kubernetes/shared-notification/README.md create mode 100644 docker/kubernetes/shared-notification/artifacts.example.json create mode 100644 docker/kubernetes/shared-notification/config.example.json create mode 100644 docker/kubernetes/shared-notification/notification.py create mode 100644 docker/kubernetes/shared-notification/prepare-seq-key.py create mode 100755 docker/kubernetes/shared-notification/prepare-storage.py create mode 100644 docker/kubernetes/shared-notification/tests/test_notification.py create mode 100644 docker/kubernetes/shared-notification/tests/test_seq_key.py diff --git a/.dockerignore b/.dockerignore index a67c691..6bda9c2 100644 --- a/.dockerignore +++ b/.dockerignore @@ -1,9 +1,10 @@ -**/.git -**/.vs -**/.idea -**/bin -**/obj -**/*.user -**/*.suo -**/TestResults -NuGet.Config \ No newline at end of file +**/.git +**/.vs +**/.idea +**/bin +**/obj +**/*.user +**/*.suo +**/TestResults +NuGet.Config +docker/kubernetes/ diff --git a/README.md b/README.md index 76c91c4..1261705 100644 --- a/README.md +++ b/README.md @@ -1,120 +1,121 @@ -# hrynco-notification-service - -Production image references now use `registry.hrynco.com`, preserving the existing image-tag variables. Contracts package repository metadata uses `gitea.hrynco.com`. These are source configuration changes; production containers and live CI configuration were not restarted or updated by this migration. - -## Documentation - -- [ItemTracker outbox email consumer](docs/itemtracker-outbox-email-consumer.md) - -## Development environment - -The development Docker Compose stack runs PostgreSQL, RabbitMQ, database migrations, -the Notification Service Web and Worker applications, Seq, and Mailpit. - -Prerequisite: install and start Docker Desktop. Then run the installation script from -the repository root: - -```powershell -.\docker\environments\install-development.cmd -``` - -The script validates Docker and the Compose configuration, builds the application -images, starts the complete stack in the background, and prints container status and -the main development URLs. It can also be launched directly from File Explorer or -from another working directory. - -To validate the setup without building images or creating containers, run: - -```powershell -.\docker\environments\install-development.cmd --validate-only -``` - -By default, the script uses the tracked `docker/environments/.env.Development` file: - -```dotenv -DB_NAME=notification_service -DB_USER=postgres -DB_PASS=postgres -VOLUME_PREFIX=ns-dev -RABBITMQ_USER=guest -RABBITMQ_PASSWORD=guest -RABBITMQ_AMQP_PORT=5672 -RABBITMQ_MANAGEMENT_PORT=15672 -WEB_PORT=5200 -DEVELOPMENT_EMAIL_SERVICE_NAME=TestService -``` - -To use personal values, copy it to the Git-ignored -`docker/environments/.env.local`, adjust the values, and run Docker Compose manually: - -```powershell -docker compose --env-file docker/environments/.env.local ` - -f docker/environments/docker-compose.yml ` - -f docker/environments/docker-compose.Development.yml ` - up --build -d -``` - -After startup, the main development endpoints are: - -- Notification Service admin: `http://localhost:5200/admin/channels` -- Notification templates: `http://localhost:5200/admin/templates` -- RabbitMQ management: `http://localhost:15672` (`guest` / `guest`) -- Mailpit inbox: `http://localhost:8025` -- Seq logs: `http://localhost:5342` -- PostgreSQL: `localhost:5433` - -During development migrations, the stack idempotently creates an active Mailpit SMTP -channel and a neutral English `TestEmail` template for -`DEVELOPMENT_EMAIL_SERVICE_NAME`. Existing channels and templates are never -overwritten. Change that environment value when the publisher uses another service -name. - -Check container status and follow Worker logs: - -```powershell -docker compose --env-file docker/environments/.env.local ` - -f docker/environments/docker-compose.yml ` - -f docker/environments/docker-compose.Development.yml ` - ps - -docker compose --env-file docker/environments/.env.local ` - -f docker/environments/docker-compose.yml ` - -f docker/environments/docker-compose.Development.yml ` - logs -f worker -``` - -Stop the environment without deleting its database and RabbitMQ volumes: - -```powershell -docker compose --env-file docker/environments/.env.local ` - -f docker/environments/docker-compose.yml ` - -f docker/environments/docker-compose.Development.yml ` - down -``` - -See the [ItemTracker consumer guide](docs/itemtracker-outbox-email-consumer.md#local-end-to-end-setup) -for the complete end-to-end setup. - -## Notification worker flow - -```mermaid -flowchart TD - A[Worker host starts] --> B[Load config and register services] - B --> C[Start SendEmailConsumer] - C --> D[Receive message from notification.send-email] - D --> E[Validate Notification.SendEmail.v1 metadata and payload] - E --> F[Resolve SendEmailService] - F --> G[Pick channel and exact-language template] - G --> H[Validate variables and render email content] - H --> I[Send via SMTP] - I --> J[Update usage counters] - J --> K[Optionally publish result to reply queue] - K --> L[Acknowledge RabbitMQ delivery] - I -. failure .-> M[Log and retry] - M -->|retries exhausted| N[Publish terminal failure result] - N --> O[Nack original delivery without requeue] -``` - -Template variables are treated as plain text. The worker HTML-encodes every variable while -rendering `HtmlBody`; subject and plain-text body interpolation preserve the original value. -Templates must express markup in `body.html` instead of supplying HTML through variables. +# hrynco-notification-service + +Production image references now use `registry.hrynco.com`, preserving the existing image-tag variables. Contracts package repository metadata uses `gitea.hrynco.com`. These are source configuration changes; production containers and live CI configuration were not restarted or updated by this migration. + +## Documentation + +- [ItemTracker outbox email consumer](docs/itemtracker-outbox-email-consumer.md) +- [Shared Notification Service Kubernetes package](docker/kubernetes/shared-notification/README.md) + +## Development environment + +The development Docker Compose stack runs PostgreSQL, RabbitMQ, database migrations, +the Notification Service Web and Worker applications, Seq, and Mailpit. + +Prerequisite: install and start Docker Desktop. Then run the installation script from +the repository root: + +```powershell +.\docker\environments\install-development.cmd +``` + +The script validates Docker and the Compose configuration, builds the application +images, starts the complete stack in the background, and prints container status and +the main development URLs. It can also be launched directly from File Explorer or +from another working directory. + +To validate the setup without building images or creating containers, run: + +```powershell +.\docker\environments\install-development.cmd --validate-only +``` + +By default, the script uses the tracked `docker/environments/.env.Development` file: + +```dotenv +DB_NAME=notification_service +DB_USER=postgres +DB_PASS=postgres +VOLUME_PREFIX=ns-dev +RABBITMQ_USER=guest +RABBITMQ_PASSWORD=guest +RABBITMQ_AMQP_PORT=5672 +RABBITMQ_MANAGEMENT_PORT=15672 +WEB_PORT=5200 +DEVELOPMENT_EMAIL_SERVICE_NAME=TestService +``` + +To use personal values, copy it to the Git-ignored +`docker/environments/.env.local`, adjust the values, and run Docker Compose manually: + +```powershell +docker compose --env-file docker/environments/.env.local ` + -f docker/environments/docker-compose.yml ` + -f docker/environments/docker-compose.Development.yml ` + up --build -d +``` + +After startup, the main development endpoints are: + +- Notification Service admin: `http://localhost:5200/admin/channels` +- Notification templates: `http://localhost:5200/admin/templates` +- RabbitMQ management: `http://localhost:15672` (`guest` / `guest`) +- Mailpit inbox: `http://localhost:8025` +- Seq logs: `http://localhost:5342` +- PostgreSQL: `localhost:5433` + +During development migrations, the stack idempotently creates an active Mailpit SMTP +channel and a neutral English `TestEmail` template for +`DEVELOPMENT_EMAIL_SERVICE_NAME`. Existing channels and templates are never +overwritten. Change that environment value when the publisher uses another service +name. + +Check container status and follow Worker logs: + +```powershell +docker compose --env-file docker/environments/.env.local ` + -f docker/environments/docker-compose.yml ` + -f docker/environments/docker-compose.Development.yml ` + ps + +docker compose --env-file docker/environments/.env.local ` + -f docker/environments/docker-compose.yml ` + -f docker/environments/docker-compose.Development.yml ` + logs -f worker +``` + +Stop the environment without deleting its database and RabbitMQ volumes: + +```powershell +docker compose --env-file docker/environments/.env.local ` + -f docker/environments/docker-compose.yml ` + -f docker/environments/docker-compose.Development.yml ` + down +``` + +See the [ItemTracker consumer guide](docs/itemtracker-outbox-email-consumer.md#local-end-to-end-setup) +for the complete end-to-end setup. + +## Notification worker flow + +```mermaid +flowchart TD + A[Worker host starts] --> B[Load config and register services] + B --> C[Start SendEmailConsumer] + C --> D[Receive message from notification.send-email] + D --> E[Validate Notification.SendEmail.v1 metadata and payload] + E --> F[Resolve SendEmailService] + F --> G[Pick channel and exact-language template] + G --> H[Validate variables and render email content] + H --> I[Send via SMTP] + I --> J[Update usage counters] + J --> K[Optionally publish result to reply queue] + K --> L[Acknowledge RabbitMQ delivery] + I -. failure .-> M[Log and retry] + M -->|retries exhausted| N[Publish terminal failure result] + N --> O[Nack original delivery without requeue] +``` + +Template variables are treated as plain text. The worker HTML-encodes every variable while +rendering `HtmlBody`; subject and plain-text body interpolation preserve the original value. +Templates must express markup in `body.html` instead of supplying HTML through variables. diff --git a/docker/kubernetes/shared-notification/.gitignore b/docker/kubernetes/shared-notification/.gitignore new file mode 100644 index 0000000..7eb50fe --- /dev/null +++ b/docker/kubernetes/shared-notification/.gitignore @@ -0,0 +1,3 @@ +*.local.json +private/ +rendered/ diff --git a/docker/kubernetes/shared-notification/README.md b/docker/kubernetes/shared-notification/README.md new file mode 100644 index 0000000..0f31205 --- /dev/null +++ b/docker/kubernetes/shared-notification/README.md @@ -0,0 +1,331 @@ +# Shared Notification Service on MicroK8s + +This package deploys one shared Web/Worker, PostgreSQL 17, RabbitMQ 4 and a temporary +Mailpit sandbox in `shared-notification`, on `mini-pc-2`. It does not expose public +Ingress, NodePort or unauthenticated administration. Application administration, +RabbitMQ management and the sandbox inbox are available through explicit local +`kubectl port-forward` sessions over SSH. + +The initial channel is `Invemory.Staging.Sandbox`, template `TestEmail`, exact language +`en`. SMTP egress permits only Mailpit:1025. Mailpit has no relay configuration. +Only its temporary inbox is ephemeral. Channels, templates, SMTP settings and usage +counters are persisted in PostgreSQL; durable broker queues, messages, users and +permissions are retained on the RabbitMQ volume. Never put real SMTP credentials +into this sandbox package. + +Mailpit runs as UID/GID `1000:1000` with dropped capabilities and writes its temporary +SQLite inbox under `/tmp` on the writable container filesystem. Exact manifest flags +and `/livez` were verified with the pinned image in an isolated local container. +Adding `readOnlyRootFilesystem` requires a separate writable `/tmp` mount; the image +exits immediately without it. The current manifest keeps that filesystem writable. + +## Verified code contract + +The current Worker consumes the fixed durable queue `notification.send-email` using +the default exchange. `ServiceName` selects channels and exact-language templates; +`CorrelationContext.ReplyTo` selects a result queue. These are logical routing +boundaries, not authorization of the message's `ServiceName`. Separate credentials +and queue permissions cannot restrict the JSON service name inside that shared queue. +There is one consumer and at-least-once processing; a redelivery can produce duplicates. +See [the consumer contract](../../../docs/itemtracker-outbox-email-consumer.md). + +This initial vhost is `/notification`. Separate broker users are generated for +administration, the Notification Worker and the future staging publisher. Worker +read permission covers only `notification.send-email`; staging reads only its result +queue. Both require write access to RabbitMQ's authorization name `amq.default` for +default-exchange publication. Resource ACLs constrain exchange/queue names, not the +routing key of a publish operation: access to the default exchange permits routing +to other existing queues within the same vhost. This MVP shared-queue design therefore +does not provide security isolation between message ServiceNames or routing keys. +No production publisher or SMTP egress is configured. +Additional environments need separately reviewed routing, users, permissions, exact +templates, provider settings and egress before they can join the shared service. + +Web/API administration has no built-in authentication in this source revision. Its +Service stays ClusterIP and default-deny ingress; access through SSH/port-forward is +restricted to operators. The worker exposes no health HTTP endpoint. Pod rollout is +a process-start check; the explicit consumer gate also verifies one broker consumer. +An actual owned smoke message and Seq event remain required after startup. + +## Storage preparation + +Owner decision on 2026-10-02: Notification PostgreSQL and RabbitMQ use the external +XFS disk. Invemory staging PostgreSQL remains on its existing internal disk. + +| Service | Data directory | Owner | Mode | Retain capacity | +| --- | --- | --- | --- | --- | +| PostgreSQL | `/mnt/shared/notification-service/postgres` (`pgdata` child) | `999:999` | `0700` | `10Gi` | +| RabbitMQ | `/mnt/shared/notification-service/rabbitmq` | `100:101` | `0700` | `5Gi` | + +`/mnt/shared` is the bind mount of `/mnt/files/shared`; both reach the same original +data. They are not backups. The required filesystem UUID is +`c2c31640-a247-4988-bc7b-9ee9789464d1`. Static local PVs use `Retain`, fixed node +affinity and prebound PVCs. Capacity declarations are scheduling/accounting budgets; +local PVs do not enforce filesystem byte quotas. + +Transfer the package to `/home/devops/notification-package`, then the owner runs +one command on mini-pc-2: + +```sh +sudo python3 /home/devops/notification-package/prepare-storage.py +``` + +The script verifies exact `/mnt/files` and `/mnt/shared` XFS mount targets and UUID +before creating anything. It accepts the already provisioned exact empty directories, +or creates missing directories under a root-owned `0755` parent. It writes root-owned +`0644` `.invemory-storage.json` markers with the UUID and roles +`shared-notification/postgres` / `shared-notification/rabbitmq`. Repeated execution +verifies markers without replacing them. It refuses symlinks, incorrect permissions, +unexpected parent contents and unmarked nonempty directories. It never deletes data. + +The deployment host preflight runs `findmnt` for each PV directory and checks actual +UUID, filesystem and ownership. Each Pod has a nonroot startup guard that checks +XFS `statfs` and the exact role/UUID marker on its mounted PVC. The marker is an +owner-provisioned assertion of UUID, not a live block-device query; the host preflight +supplies that check. A missing external mount falls onto ext4 and fails the guard. +These checks and `Retain` do not protect against direct manual deletion. + +Storage Pods deliberately omit `fsGroup` so kubelet does not recursively change +the owner-provisioned directory and marker permissions. PostgreSQL uses the pinned +image's writable `/var/run/postgresql` directory on its container filesystem; +RabbitMQ writes its cookie/state in its UID-owned retained data directory. Both run +with explicit nonroot UID/GID, dropped capabilities and restricted Pod security. + +RabbitMQ diagnostics probes allow 15 seconds per call: a healthy call on the +resource-limited node took seven seconds. Startup keeps a three-minute failure +budget (nine attempts, every 20 seconds); readiness runs every 20 seconds and +liveness every 30 seconds. Startup must succeed before the other probes run. + +Never remove files directly from these data directories. Storage path/UUID/node/ +capacity changes and database/broker image changes are rejected by repeated deploys. +A future storage move requires a separate controlled procedure: stop publishers and +both service writers, verify database and broker backups, stop storage processes, +copy and verify offline data, provision target markers/ownership, bind new retained +claims, test restoration and perform rollback checks before retiring old storage. +This package does not perform that migration. + +## Artifacts and private state + +Copy `config.example.json` to `config.local.json` and verify actual Seq/publisher Pod +selectors. The example reply queue `invemory.staging.notification-results` is isolated +from the old client fallback. Its exact name must be agreed and configured in the +Invemory publisher/result consumer before that later cutover. The current publisher +remains `LogOnly`, and its result consumer remains disabled. + +Build Web, Worker and Migrator from the same verified source archive, using each +existing Dockerfile. Record the checked commit and archive SHA256, including any +explicitly recorded uncommitted delta. Mirror PostgreSQL, RabbitMQ, Mailpit and a +Python 3 Linux image to `registry.hrynco.com`. Copy `artifacts.example.json` to +`artifacts.local.json` and record all seven immutable image digests, source hashes +and the two reviewed migration IDs. The script rejects placeholder/tag references, +other registries, mismatched application/migrator archives and unsupported DB/broker +major versions. The archive hash fields are an operator provenance record; independently +verify the build inputs and registry digests before live application. + +Provision a **dedicated** Seq ingestion key for Notification Service, with its own +identity/permissions, in a private file. Do not reuse the staging application's stamped +key. Store registry authentication as a private Docker `config.json` with only +`registry.hrynco.com` in `auths`. Never print key values, full Secrets or credentials. + +Run the dedicated helper as devops on mini-pc-2 after package review: + +```sh +python3 /home/devops/notification-package/prepare-seq-key.py +``` + +It reads the existing private `/home/devops/seq-first/credentials.json` silently and +uses the checked session/CSRF API pattern at the trusted Seq LAN endpoint. It creates +only `HrynCo Notification Service`, granting only `Ingest` and stamping +`Environment=SharedNotification`, `Service=HrynCo.NotificationService`. `Application` +remains supplied by each application. It stores the token only in +`/home/devops/notification-private/seq-key.txt` (`0600`, parent `0700`) before the +create request, so an unknown POST outcome can be retried with the same token. +Repeats verify existing permissions/properties and preserve the token; mismatches, +duplicate keys or a missing local token for an existing key fail without rotation. +The installed Seq does not expose the original token after creation, and its +EventEntity has no `ApiKeyId`. The helper therefore adds the server-applied property +`NotificationIngestKeyId=` through a guarded PUT of **only** the +dedicated key (`apikey-79` at the initial checkpoint). It preserves its token, +permissions, all other input settings and entity fields. It verifies that no other +key has the same identity marker before and after updating using explicit +`api/apikeys?shared=true` lists (including shared keys), and verifies exact +readback. Repeats with the correct marker perform no PUT. An unknown PUT outcome is +recovered by reading the existing marker on the next run, without token rotation or +another key. A marker mismatch/collision stops without changing another key. +The [official Seq API update contract](https://github.com/datalust/seq-api/blob/dev/src/Seq.Api/ResourceGroups/ApiKeysResourceGroup.cs) +states that updating settings cannot change the API key token. +HTTP redirects are denied before forwarding credentials; environment proxies are +disabled for this fixed LAN operation. Each run verifies token ingestion using one +technical canary event without user payload or credentials, then reads that exact +`ProvisioningCheckId` through the admin session with bounded polling. It verifies +the actual server-stamped `NotificationIngestKeyId`, Environment and Service. The +canary body omits all three stamps, so a valid staging/foreign token cannot supply +the dedicated identity marker. If `ApiKeyId` is exposed by a server version, it must +also agree. Missing or conflicting identity proof fails closed without rotation. +No event payload/token is printed. Back up the private file. + +The key's stamped properties take precedence over application fields. Verify the first +Web/Worker event has `Environment=SharedNotification` before accepting delivery. +The helper never changes staging keys, signals, retention or Seq deployment. +The existing shared Seq retention/storage policy remains in force; changing retention +is a separate observability decision. Dedicated key and event verification remain +mandatory live gates if the helper has not yet been run. + +`bootstrap` generates random passwords once in +`/home/devops/shared-notification-private/credentials.json` (`0600`, parent `0700`) +and reconciles Kubernetes Secrets. Keep and back up this private file outside Git. +Repeats reuse it, compare live Secret data and refuse implicit password/key/routing +rotation. If the file is missing while a live Secret exists, recover the original +file; generating new credentials would not change an existing PostgreSQL role. +Registry authentication can be refreshed independently. `.local.json`, rendered files +and `private/` are ignored in this package, but prefer keeping secrets outside the repo. + +## Ordered installation and repeat deployment + +All mutating commands run as devops on mini-pc-2. `prepare-storage.py` is the only +owner-run root operation. Scripts use Python's standard library, MicroK8s kubectl, +`/home/devops/.kube/cluster-admin.conf` and bounded timeouts. No command prints Secrets. + +1. Review source changes, artifacts and manifests independently. Measure live node + requests/limits and usage; inspect namespace/node pressure. The namespace budget is + 8 Pods, 5 Deployments, 3 Jobs, 2 PVCs, 15Gi storage, requests1600m/2Gi and limits4CPU/4Gi. + This is a separate budget; existing `invemory-staging` quotas are not expanded. +2. Run owner storage preparation and privately supply registry/Seq files. +3. Bootstrap the namespace and preserved Secrets: + +```sh +python3 notification.py bootstrap --config config.local.json --artifacts artifacts.local.json --seq-key-file /home/devops/notification-private/seq-key.txt --registry-auth-file /home/devops/notification-private/registry-auth.json +``` + +4. Render for review and run server-side dry-run. `check` also rejects resource + adoption, wrong mounted disk, Pod selector mismatches and storage/image changes: + +```sh +python3 notification.py render --config config.local.json --artifacts artifacts.local.json --output /home/devops/notification-manifests.json +python3 notification.py check --config config.local.json --artifacts artifacts.local.json +``` + +5. First install applies storage/isolated foundation and runs migrations with Web and + Worker stopped. The migration target is exactly `InitialCreate` + `PendingChanges`. + After migration success, the script verifies history, grants a separate + `notification_runtime` role only DML on the three application tables, and idempotently + seeds the Mailpit channel and neutral English `TestEmail`. Existing matching-service + channels/templates are left unchanged. It resumes Web after these gates; Worker + remains stopped for inspection: + +```sh +python3 notification.py deploy --config config.local.json --artifacts artifacts.local.json --migrate +``` + +6. Inspect channel/template, queue backlog/permissions, DB role and Seq identity. + Start the Worker only after those checks. The script verifies the sandbox SMTP + settings, required template, empty incoming queue and exactly one consumer: + +```sh +python3 notification.py deploy --config config.local.json --artifacts artifacts.local.json --enable-worker --sandbox-verified +``` + +7. Send one owned neutral test message to Mailpit; inspect its matching result, + usage counter, ACK and redacted technical logs. Test malformed metadata and + missing template/variable failure with owned safe fixtures. Remove only owned + test data through application/broker APIs. No arbitrary live email or client backlog + is authorized by this smoke test. + +Ordinary repeat `deploy` verifies target schema and preserves the current Worker +replica count. It reapplies canonical manifests with the same persistent claims and +Secret references. When Worker remains enabled, it waits for its rollout and verifies +exactly one broker consumer before reporting completion. Both explicit enable and +ordinary redeploy allow up to 60 seconds after rollout for a zero consumer count to +become one, polling with a two-second pause and bounded broker commands. Extra or +unreadable consumer counts fail immediately. Rollout failure, verification failure +or timeout scales Worker to zero and stops the deployment without reporting success. +It does not recreate Secrets +or reseed provider settings. A later +`--migrate` stops Web/Worker, takes a verified local PostgreSQL custom-format dump, +runs the Migrator and history/grant/seed gates, then resumes Web. Worker stays stopped +unless `--enable-worker --sandbox-verified` is supplied and all gates pass. Once +writers have been stopped, backup, migration, history or grants failure leaves them +stopped. An initial history mismatch on an already ready database is rejected before +scaling writers, without changing their state. Failed Jobs remain for +private inspection; the next run replaces only its own named Job. Migration locking +uses a private host file and is local to this deployment host. + +An interrupted first install can leave the PostgreSQL Deployment present with no +ready Pod (for example, before owner storage markers exist). A retry with `--migrate` +first verifies ownership/storage/images, stops both writers and reconciles only the +foundation/storage workloads. It waits for PostgreSQL, RabbitMQ and Mailpit readiness +before querying actual migration history. Every migration cycle backs up the actual +initialized database, including an empty first database: Deployment absence/existence +never establishes that the retained volume has no data. Backup/history/storage +failure keeps writers stopped and prevents migration. An ordinary deploy against an +unready database requires this explicit recovery path and performs no mutation. +Stopping an already empty Deployment is supported: collection reads require a JSON +PodList, and an empty `items` list needs no Pod-termination wait. Named resource +lookups alone use `--ignore-not-found`. + +Backups are retained under the private directory with a SHA256/byte-count/history +manifest. `pg_restore --list` validates dump structure; a full restore test and offsite +copy are separate gates. This backup covers the Notification database only. It is not +a snapshot of RabbitMQ's queue files. Migration does not upgrade/recreate the broker +volume; durable queues remain on that retained volume. Broker disk backup/recovery +must be verified before a broker upgrade or storage migration. + +## Access and networking + +Use SSH to the deployment host, then forward one local port when needed: + +```sh +/snap/microk8s/current/kubectl --kubeconfig=/home/devops/.kube/cluster-admin.conf -n shared-notification port-forward service/web 5200:8080 --address=127.0.0.1 +``` + +Apply equivalent explicit forwards for `service/rabbitmq`15672 and `service/mailpit`8025. +Keep them bound to loopback. ClusterIP services expose no LAN/public ports. + +Default-deny policies allow namespace-scoped DB5432, shared Worker/bounded staging +publisher AMQP5672, sandbox SMTP1025, cluster DNS53 and Web/Worker Seq8080 only. +The package adds its own `notification-seq-ingress` policy in `observability` and +`notification-broker-egress` in `invemory-staging`; existing policies stay owned by +their original packages. Verify these exact selectors in the rendered diff. +No SMTP internet egress or broad namespace allowance is included. + +The Invemory publisher/result-consumer cutover remains a separate operation after +Notification acceptance, exact staged settings and backlog inspection. The package +does not change Invemory configuration, production routes or CI deployment. + +## Local validation + +```powershell +python -m unittest discover -s docker/kubernetes/shared-notification/tests -v +dotnet build HrynCo.NotificationService.slnx --no-restore +dotnet test HrynCo.NotificationService.slnx --no-build --no-restore +``` + +Unit tests cover immutable artifacts, Secret preservation structure, isolated network +and storage contracts, failure ordering and no automatic Worker activation. Passing +local tests does not establish live storage, image startup, Seq access, mail delivery, +backups/restoration or repeat-deploy acceptance; record these after actual testing. + +## Verified mini-pc-2 sandbox checkpoint (2026-10-02) + +The initial installation from source commit `a44b98292f9001f28fed425be131a2c993a85847` +completed on mini-pc-2. The external XFS storage guards passed; PostgreSQL, RabbitMQ, +Mailpit, Web and one Worker are Ready with zero restarts. The namespace has no Ingress +or NodePort. `InitialCreate` and `PendingChanges`, runtime grants and sandbox seed +completed. Two private local PostgreSQL backups were created and validated by the +deployment script before the initial and repeated migration runs. Neither backup is +an offsite copy or a RabbitMQ backup; full restoration was not tested. + +The dedicated Seq key was verified on live ingestion and Web/Worker events carry +`Environment=SharedNotification`. A single owned `TestEmail` message reached only +Mailpit, was ACKed, produced a matching successful result and incremented the usage +counter to one. An unsupported message type was rejected without an email; a missing +template produced a matching negative result without an email. The owned messages and +results were removed afterward. Both queues are empty and the incoming queue has one +consumer. Ordinary repeat deploy with the Worker running and a repeated no-op +migration passed; the Worker was enabled again afterward. The external volume data +measured about 47MiB for PostgreSQL and 292KiB for RabbitMQ at this checkpoint. + +Invemory staging still publishes Outbox entries in `LogOnly` mode and does not +consume results. Connecting its publisher/result consumer and proving a real staging +application flow remain separate gated work. Other environments and production SMTP +are not part of this sandbox checkpoint. diff --git a/docker/kubernetes/shared-notification/artifacts.example.json b/docker/kubernetes/shared-notification/artifacts.example.json new file mode 100644 index 0000000..0caa16f --- /dev/null +++ b/docker/kubernetes/shared-notification/artifacts.example.json @@ -0,0 +1,14 @@ +{ + "sourceCommit": "REPLACE_WITH_CHECKED_SOURCE_COMMIT", + "sourceArchiveSha256": "REPLACE_WITH_VERIFIED_SOURCE_ARCHIVE_SHA256", + "migrations": ["20260501214629_InitialCreate", "20260502154249_PendingChanges"], + "images": { + "web": {"image": "registry.hrynco.com/hrynco.notificationservice.web@sha256:REPLACE", "sourceArchiveSha256": "REPLACE_WITH_VERIFIED_SOURCE_ARCHIVE_SHA256"}, + "worker": {"image": "registry.hrynco.com/hrynco.notificationservice.worker@sha256:REPLACE", "sourceArchiveSha256": "REPLACE_WITH_VERIFIED_SOURCE_ARCHIVE_SHA256"}, + "migrator": {"image": "registry.hrynco.com/hrynco.notificationservice.migrator@sha256:REPLACE", "sourceArchiveSha256": "REPLACE_WITH_VERIFIED_SOURCE_ARCHIVE_SHA256"}, + "postgres": {"image": "registry.hrynco.com/mirror/postgres@sha256:REPLACE", "version": "17"}, + "rabbitmq": {"image": "registry.hrynco.com/mirror/rabbitmq@sha256:REPLACE", "version": "4"}, + "mailpit": {"image": "registry.hrynco.com/mirror/mailpit@sha256:REPLACE"}, + "guard": {"image": "registry.hrynco.com/mirror/python@sha256:REPLACE"} + } +} diff --git a/docker/kubernetes/shared-notification/config.example.json b/docker/kubernetes/shared-notification/config.example.json new file mode 100644 index 0000000..8bd83b7 --- /dev/null +++ b/docker/kubernetes/shared-notification/config.example.json @@ -0,0 +1,19 @@ +{ + "namespace": "shared-notification", + "node": "mini-pc-2", + "storage": { + "basePath": "/mnt/shared/notification-service", + "filesystem": "xfs", + "uuid": "c2c31640-a247-4988-bc7b-9ee9789464d1", + "postgresSize": "10Gi", + "rabbitmqSize": "5Gi" + }, + "sandboxServiceName": "Invemory.Staging.Sandbox", + "rabbitVirtualHost": "/notification", + "replyQueue": "invemory.staging.notification-results", + "publisherNamespace": "invemory-staging", + "publisherPodLabels": {"app.kubernetes.io/component": "worker"}, + "seqNamespace": "observability", + "seqPodLabels": {"app.kubernetes.io/name": "seq"}, + "seqUrl": "http://seq.observability.svc.cluster.local:8080" +} diff --git a/docker/kubernetes/shared-notification/notification.py b/docker/kubernetes/shared-notification/notification.py new file mode 100644 index 0000000..de3b1b6 --- /dev/null +++ b/docker/kubernetes/shared-notification/notification.py @@ -0,0 +1,695 @@ +#!/usr/bin/env python3 +"""Bounded shared Notification Service deployment. Python standard library only.""" + +import argparse +import base64 +import contextlib +import datetime +import hashlib +import json +import os +from pathlib import Path, PurePosixPath +import re +import secrets +import subprocess +import sys +import time + +NAMESPACE = "shared-notification" +MANAGED = "hrynco-shared-notification" +MIGRATIONS = ["20260501214629_InitialCreate", "20260502154249_PendingChanges"] +IMAGE = re.compile(r"registry\.hrynco\.com/[a-z0-9._/-]+@sha256:[a-f0-9]{64}$") +SHA = re.compile(r"[a-f0-9]{64}$") + +# Marker must be provisioned only after findmnt proves the backing filesystem UUID. +# XFS/ext4 statfs plus exact marker prevents a missing-mount fallback onto another FS. +# Marker alone is not a live query of the block-device UUID: verify findmnt at preflight. +GUARD = r''' +import ctypes, json, os +buf = ctypes.create_string_buffer(512) +libc = ctypes.CDLL(None, use_errno=True) +if libc.statfs(b'/data', ctypes.byref(buf)) != 0: + raise RuntimeError('storage statfs failed') +magic = ctypes.cast(buf, ctypes.POINTER(ctypes.c_long))[0] +expected = {'ext4': 0xEF53, 'xfs': 0x58465342}[os.environ['EXPECTED_FS']] +if magic != expected: + raise RuntimeError('unexpected storage filesystem') +with open('/data/.invemory-storage.json', encoding='utf-8') as stream: + marker = json.load(stream) +if marker != {'uuid': os.environ['EXPECTED_UUID'], 'role': os.environ['EXPECTED_ROLE']}: + raise RuntimeError('unexpected storage UUID/role marker') +print('storage guard passed') +''' + +# Variables are quoted by PostgreSQL format, never interpolated into SQL by the shell. +GRANTS = r''' +\getenv runtime_password RUNTIME_PASSWORD +SELECT format('CREATE ROLE notification_runtime LOGIN PASSWORD %L NOSUPERUSER NOCREATEDB NOCREATEROLE', :'runtime_password') +WHERE NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'notification_runtime') \gexec +SELECT format('ALTER ROLE notification_runtime PASSWORD %L', :'runtime_password') \gexec +REVOKE ALL ON DATABASE notification_service FROM PUBLIC; +GRANT CONNECT ON DATABASE notification_service TO notification_runtime; +REVOKE CREATE ON SCHEMA public FROM PUBLIC; +GRANT USAGE ON SCHEMA public TO notification_runtime; +GRANT SELECT, INSERT, UPDATE, DELETE ON email_channels, email_templates, email_channel_usage TO notification_runtime; +''' + +SANDBOX = r''' +BEGIN; +INSERT INTO email_channels (id, service_name, priority, provider_type, settings, warn_threshold_percent, is_active, created) +SELECT gen_random_uuid(), 'Invemory.Staging.Sandbox', 1, 1, + '{"EmailChannelType":1,"Host":"mailpit","Port":1025,"Username":"","Password":"","UseSsl":false,"FromEmail":"no-reply@local.hrynco.test","FromName":"Invemory Sandbox"}'::jsonb, 90, true, now() +WHERE NOT EXISTS (SELECT 1 FROM email_channels WHERE service_name = 'Invemory.Staging.Sandbox'); +INSERT INTO email_templates (id, service_name, key, language_code, subject, html_body, text_body, variables, created) +VALUES (gen_random_uuid(), 'Invemory.Staging.Sandbox', 'TestEmail', 'en', 'Test notification', + '

Hello {{RecipientName}},

{{Message}}

', 'Hello {{RecipientName}}, {{Message}}', + '[{"name":"RecipientName","required":true},{"name":"Message","required":true}]'::jsonb, now()) +ON CONFLICT (service_name, key, language_code) DO NOTHING; +COMMIT; +''' + +SANDBOX_CHECK = """SELECT + (SELECT count(*) = 1 AND bool_and(provider_type = 1 AND is_active AND + settings->>'Host' = 'mailpit' AND settings->>'Port' = '1025' AND + settings->>'UseSsl' = 'false' AND settings->>'Username' = '' AND settings->>'Password' = '') + FROM email_channels WHERE service_name = 'Invemory.Staging.Sandbox') AND + EXISTS (SELECT 1 FROM email_templates WHERE service_name = 'Invemory.Staging.Sandbox' + AND key = 'TestEmail' AND language_code = 'en' + AND variables @> '[{"name":"RecipientName","required":true},{"name":"Message","required":true}]'::jsonb)""" + + +def read_json(path): + return json.loads(Path(path).read_text(encoding="utf-8")) + + +def write_private(path, value): + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(value, stream, indent=2) + stream.write("\n") + + +def validate(config, artifacts): + if config.get("namespace") != NAMESPACE: + raise ValueError("this package manages only shared-notification") + if not re.fullmatch(r"[a-z0-9][a-z0-9.-]*", config["node"]): + raise ValueError("invalid node name") + storage = config["storage"] + path = PurePosixPath(storage["basePath"]) + if not path.is_absolute() or ".." in path.parts or str(path) != storage["basePath"]: + raise ValueError("storage must be an absolute normalized path") + if str(path) not in ("/home/devops/shared-notification", "/mnt/shared/notification-service"): + raise ValueError("unexpected storage path; a new location requires package review") + if storage["filesystem"] not in ("ext4", "xfs"): + raise ValueError("unsupported filesystem") + if not re.fullmatch(r"[a-f0-9-]{36}", storage["uuid"]): + raise ValueError("invalid storage UUID") + for name in ("postgresSize", "rabbitmqSize"): + if not re.fullmatch(r"[1-9][0-9]*Gi", storage[name]): + raise ValueError("invalid storage capacity") + for key in ("sandboxServiceName", "replyQueue", "rabbitVirtualHost"): + if not re.fullmatch(r"[A-Za-z0-9._/-]{1,100}", config[key]): + raise ValueError("invalid routing value") + if config["sandboxServiceName"] != "Invemory.Staging.Sandbox": + raise ValueError("initial package is restricted to the reviewed sandbox identity") + if config["seqNamespace"] != "observability" or config["publisherNamespace"] != "invemory-staging": + raise ValueError("unexpected dependent namespace") + if config["seqUrl"] != "http://seq.observability.svc.cluster.local:8080": + raise ValueError("unexpected Seq endpoint") + for key in ("seqPodLabels", "publisherPodLabels"): + if not isinstance(config[key], dict) or not config[key]: + raise ValueError("exact dependent pod selector required") + if not re.fullmatch(r"[a-f0-9]{40}", artifacts.get("sourceCommit", "")): + raise ValueError("checked source commit required") + if not SHA.fullmatch(artifacts.get("sourceArchiveSha256", "")): + raise ValueError("verified source archive hash required") + if artifacts.get("migrations") != MIGRATIONS: + raise ValueError("migration list differs from reviewed source") + expected = {"web", "worker", "migrator", "postgres", "rabbitmq", "mailpit", "guard"} + if set(artifacts["images"]) != expected: + raise ValueError("complete image set required") + for name, artifact in artifacts["images"].items(): + if not IMAGE.fullmatch(artifact["image"]) or artifact["image"].endswith("0" * 64): + raise ValueError("registry.hrynco.com immutable image required: " + name) + if name in ("web", "worker", "migrator") and artifact.get("sourceArchiveSha256") != artifacts["sourceArchiveSha256"]: + raise ValueError("application and migrator source archives differ") + if artifacts["images"]["postgres"].get("version") != "17" or artifacts["images"]["rabbitmq"].get("version") != "4": + raise ValueError("database/broker major upgrade requires separate reviewed procedure") + + +def labels(component=None): + result = {"app.kubernetes.io/name": MANAGED, "app.kubernetes.io/managed-by": MANAGED} + if component: + result["app.kubernetes.io/component"] = component + return result + + +def obj(kind, name, spec=None, namespace=NAMESPACE, api="v1", **extra): + value = {"apiVersion": api, "kind": kind, "metadata": {"name": name, "labels": labels()}} + if namespace: + value["metadata"]["namespace"] = namespace + if spec is not None: + value["spec"] = spec + value.update(extra) + return value + + +def env(name, value=None, secret=None, key=None): + if secret: + return {"name": name, "valueFrom": {"secretKeyRef": {"name": secret, "key": key}}} + return {"name": name, "value": str(value)} + + +def secure(uid=1000, gid=1000): + return {"runAsNonRoot": True, "runAsUser": uid, "runAsGroup": gid, + "allowPrivilegeEscalation": False, "capabilities": {"drop": ["ALL"]}, + "seccompProfile": {"type": "RuntimeDefault"}} + + +def manifest(config, artifacts): + validate(config, artifacts) + images = {key: value["image"] for key, value in artifacts["images"].items()} + storage = config["storage"] + output = [obj("Namespace", NAMESPACE, namespace=None)] + output[0]["metadata"]["labels"].update({"pod-security.kubernetes.io/enforce": "restricted"}) + output.append(obj("ResourceQuota", "notification-budget", {"hard": { + "pods": "8", "count/deployments.apps": "5", "count/jobs.batch": "3", + "persistentvolumeclaims": "2", "requests.storage": str(int(storage["postgresSize"][:-2]) + int(storage["rabbitmqSize"][:-2])) + "Gi", + "requests.cpu": "1600m", "requests.memory": "2Gi", "limits.cpu": "4", "limits.memory": "4Gi"}})) + output.append(obj("LimitRange", "container-budget", {"limits": [{"type": "Container", + "defaultRequest": {"cpu": "50m", "memory": "64Mi"}, "default": {"cpu": "250m", "memory": "256Mi"}}]})) + for component, size, uid, gid in (("postgres", storage["postgresSize"], 999, 999), ("rabbitmq", storage["rabbitmqSize"], 100, 101)): + pv = obj("PersistentVolume", "shared-notification-" + component, { + "capacity": {"storage": size}, "accessModes": ["ReadWriteOnce"], "volumeMode": "Filesystem", + "persistentVolumeReclaimPolicy": "Retain", "storageClassName": "", + "claimRef": {"namespace": NAMESPACE, "name": component + "-data"}, + "local": {"path": storage["basePath"] + "/" + component}, + "nodeAffinity": {"required": {"nodeSelectorTerms": [{"matchExpressions": [{ + "key": "kubernetes.io/hostname", "operator": "In", "values": [config["node"]]}]}]}}}, namespace=None) + pv["metadata"]["annotations"] = {"hrynco.com/storage-uuid": storage["uuid"], "hrynco.com/filesystem": storage["filesystem"]} + output += [pv, obj("PersistentVolumeClaim", component + "-data", { + "accessModes": ["ReadWriteOnce"], "volumeMode": "Filesystem", "storageClassName": "", + "volumeName": pv["metadata"]["name"], "resources": {"requests": {"storage": size}}})] + output.append(obj("ConfigMap", "rabbit-config", data={"rabbitmq.conf": + "listeners.tcp.default = 5672\nmanagement.tcp.port = 15672\nmanagement.load_definitions = /etc/rabbitmq/definitions/definitions.json\n"})) + output.append(obj("ConfigMap", "runtime-grants", data={"grants.sql": GRANTS})) + output.append(obj("ConfigMap", "sandbox-seed", data={"sandbox.sql": SANDBOX})) + output.append(obj("ConfigMap", "notification-config", data={ + "ASPNETCORE_ENVIRONMENT": "Production", "DOTNET_ENVIRONMENT": "Production", + "ASPNETCORE_HTTP_PORTS": "8080", "App__RabbitMq__Host": "rabbitmq", + "App__RabbitMq__Port": "5672", "App__RabbitMq__VirtualHost": config["rabbitVirtualHost"], + "Serilog__WriteTo__1__Name": "Seq", "Serilog__WriteTo__1__Args__serverUrl": config["seqUrl"], + "Serilog__Properties__Environment": "Shared", "Serilog__Properties__Service": "HrynCo.NotificationService"})) + + def workload(component, image, resources, environment=None, ports=None, volumes=None, mounts=None, replicas=1, uid=1000, gid=1000, **extra): + container = {"name": component, "image": image, "imagePullPolicy": "IfNotPresent", + "resources": resources, "securityContext": secure(uid, gid)} + if environment: + container["env"] = environment + if ports: + container["ports"] = [{"name": name, "containerPort": port} for name, port in ports] + if mounts: + container["volumeMounts"] = mounts + container.update(extra) + pod = {"automountServiceAccountToken": False, "nodeSelector": {"kubernetes.io/hostname": config["node"]}, + "imagePullSecrets": [{"name": "registry-pull"}], "containers": [container], + "terminationGracePeriodSeconds": 60} + if volumes: + pod["volumes"] = volumes + if component in ("postgres", "rabbitmq"): + pod["initContainers"] = [{"name": "storage-guard", "image": images["guard"], + "command": ["python3", "-c", GUARD], "securityContext": secure(uid, gid), + "resources": {"requests": {"cpu": "10m", "memory": "32Mi"}, "limits": {"cpu": "100m", "memory": "64Mi"}}, + "env": [env("EXPECTED_UUID", storage["uuid"]), env("EXPECTED_ROLE", NAMESPACE + "/" + component), env("EXPECTED_FS", storage["filesystem"])], + "volumeMounts": [{"name": "data", "mountPath": "/data", "readOnly": True}]}] + value = obj("Deployment", component, {"replicas": replicas, "strategy": {"type": "Recreate"}, + "selector": {"matchLabels": labels(component)}, "template": {"metadata": {"labels": labels(component), + "annotations": {"hrynco.com/source-archive-sha256": artifacts["sourceArchiveSha256"]}}, "spec": pod}}, api="apps/v1") + output.append(value) + if ports: + output.append(obj("Service", component, {"selector": labels(component), "ports": [ + {"name": name, "port": port, "targetPort": name} for name, port in ports]})) + + def res(cpu, memory, limit_cpu, limit_memory): + return {"requests": {"cpu": cpu, "memory": memory}, "limits": {"cpu": limit_cpu, "memory": limit_memory}} + + workload("postgres", images["postgres"], res("100m", "256Mi", "500m", "512Mi"), + [env("POSTGRES_DB", "notification_service"), env("POSTGRES_USER", "notification_owner"), + env("POSTGRES_PASSWORD", secret="notification-private", key="postgresPassword"), env("PGDATA", "/data/pgdata")], + [("postgres", 5432)], [{"name": "data", "persistentVolumeClaim": {"claimName": "postgres-data"}}], + [{"name": "data", "mountPath": "/data"}], uid=999, gid=999, + readinessProbe={"exec": {"command": ["pg_isready", "-U", "notification_owner", "-d", "notification_service"]}, "periodSeconds": 5}, + livenessProbe={"exec": {"command": ["pg_isready", "-U", "notification_owner", "-d", "notification_service"]}, "initialDelaySeconds": 30, "periodSeconds": 10}) + workload("rabbitmq", images["rabbitmq"], res("100m", "256Mi", "500m", "512Mi"), + [env("RABBITMQ_NODENAME", "rabbit@rabbitmq"), env("RABBITMQ_DEFAULT_USER", "notification_admin"), + env("RABBITMQ_DEFAULT_PASS", secret="notification-private", key="rabbitAdminPassword")], + [("amqp", 5672), ("management", 15672)], + [{"name": "data", "persistentVolumeClaim": {"claimName": "rabbitmq-data"}}, + {"name": "config", "configMap": {"name": "rabbit-config"}}, {"name": "definitions", "secret": {"secretName": "rabbit-definitions"}}], + [{"name": "data", "mountPath": "/var/lib/rabbitmq"}, {"name": "config", "mountPath": "/etc/rabbitmq/rabbitmq.conf", "subPath": "rabbitmq.conf", "readOnly": True}, + {"name": "definitions", "mountPath": "/etc/rabbitmq/definitions", "readOnly": True}], uid=100, gid=101, + startupProbe={"exec": {"command": ["rabbitmq-diagnostics", "-q", "ping"]}, "timeoutSeconds": 15, "periodSeconds": 20, "failureThreshold": 9}, + readinessProbe={"exec": {"command": ["rabbitmq-diagnostics", "-q", "check_running"]}, "timeoutSeconds": 15, "periodSeconds": 20}, + livenessProbe={"exec": {"command": ["rabbitmq-diagnostics", "-q", "ping"]}, "timeoutSeconds": 15, "periodSeconds": 30}) + output[-2]["spec"]["template"]["spec"]["hostname"] = "rabbitmq" + workload("mailpit", images["mailpit"], res("25m", "32Mi", "100m", "128Mi"), + [env("MP_SMTP_BIND_ADDR", "0.0.0.0:1025"), env("MP_UI_BIND_ADDR", "0.0.0.0:8025"), env("MP_MAX_MESSAGES", "100")], + [("smtp", 1025), ("http", 8025)], readinessProbe={"httpGet": {"path": "/livez", "port": "http"}, "periodSeconds": 10}) + for component in ("web", "worker"): + environment = [env("App__ConnectionString", secret="notification-private", key="runtimeConnectionString"), + env("Serilog__WriteTo__1__Args__apiKey", secret="notification-private", key="seqApiKey")] + if component == "worker": + environment += [env("App__RabbitMq__User", "notification_worker"), env("App__RabbitMq__Password", secret="notification-private", key="rabbitWorkerPassword")] + workload(component, images[component], res("50m", "128Mi", "250m", "256Mi"), environment, + [("http", 8080)] if component == "web" else None, replicas=0, + **({"readinessProbe": {"httpGet": {"path": "/api/v1/email-channels?serviceName=" + config["sandboxServiceName"], "port": "http"}, "periodSeconds": 10}} if component == "web" else {})) + deployment = output[-2] if component == "web" else output[-1] + deployment["spec"]["template"]["spec"]["containers"][0]["envFrom"] = [{"configMapRef": {"name": "notification-config"}}] + output += network_policies(config) + return output + + +def peer(component=None, namespace=NAMESPACE, selector=None): + return {"namespaceSelector": {"matchLabels": {"kubernetes.io/metadata.name": namespace}}, + "podSelector": {"matchLabels": selector if selector is not None else labels(component)}} + + +def network_policies(config): + def policy(name, selector, ingress=None, egress=None, namespace=NAMESPACE): + spec = {"podSelector": {"matchLabels": selector}, "policyTypes": []} + if ingress is not None: + spec.update(ingress=ingress) + spec["policyTypes"].append("Ingress") + if egress is not None: + spec.update(egress=egress) + spec["policyTypes"].append("Egress") + return obj("NetworkPolicy", name, spec, namespace, "networking.k8s.io/v1") + + def port(number, protocol="TCP"): + return {"protocol": protocol, "port": number} + + output = [policy("default-deny", {}, [], [])] + output.append(policy("dns", labels(), egress=[{"to": [peer(namespace="kube-system", selector={"k8s-app": "kube-dns"})], "ports": [port(53, "UDP"), port(53)]}])) + output.append(policy("postgres-ingress", labels("postgres"), ingress=[{"from": [peer(x) for x in ("web", "worker", "migrator", "grants", "sandbox")], "ports": [port(5432)]}])) + output.append(policy("rabbitmq-ingress", labels("rabbitmq"), ingress=[{"from": [peer("worker"), peer(namespace=config["publisherNamespace"], selector=config["publisherPodLabels"])], "ports": [port(5672)]}])) + output.append(policy("mailpit-ingress", labels("mailpit"), ingress=[{"from": [peer("worker"), peer("web")], "ports": [port(1025)]}])) + for component in ("web", "worker", "migrator", "grants", "sandbox"): + rules = [{"to": [peer("postgres")], "ports": [port(5432)]}] + if component in ("web", "worker"): + rules.append({"to": [peer(namespace=config["seqNamespace"], selector=config["seqPodLabels"])], "ports": [port(8080)]}) + if component == "worker": + rules += [{"to": [peer("rabbitmq")], "ports": [port(5672)]}, {"to": [peer("mailpit")], "ports": [port(1025)]}] + if component == "web": + rules.append({"to": [peer("mailpit")], "ports": [port(1025)]}) + output.append(policy(component + "-egress", labels(component), egress=rules)) + # Additive allowances, separately owned; existing deny policies remain effective. + output.append(policy("notification-seq-ingress", config["seqPodLabels"], ingress=[{ + "from": [peer("web"), peer("worker")], "ports": [port(8080)]}], namespace=config["seqNamespace"])) + # Publisher remains LogOnly until its separate, explicitly controlled cutover. + output.append(policy("notification-broker-egress", config["publisherPodLabels"], egress=[{ + "to": [peer("rabbitmq")], "ports": [port(5672)]}], namespace=config["publisherNamespace"])) + return output + + +def rabbit_definitions(config, credentials): + def hashed(password, salt): + return base64.b64encode(salt + hashlib.sha256(salt + password.encode()).digest()).decode() + + users = [] + for name, key, tag in (("notification_admin", "rabbitAdminPassword", "administrator"), ("notification_worker", "rabbitWorkerPassword", ""), ("invemory_staging", "rabbitPublisherPassword", "")): + salt = base64.b64decode(credentials[key + "Salt"]) + users.append({"name": name, "password_hash": hashed(credentials[key], salt), "hashing_algorithm": "rabbit_password_hashing_sha256", "tags": tag}) + incoming = re.escape("notification.send-email") + reply = re.escape(config["replyQueue"]) + vhost = config["rabbitVirtualHost"] + return {"users": users, "vhosts": [{"name": vhost}], "permissions": [ + {"user": "notification_admin", "vhost": vhost, "configure": ".*", "write": ".*", "read": ".*"}, + {"user": "notification_worker", "vhost": vhost, "configure": "^(" + incoming + "|" + reply + ")$", "write": "^(amq\\.default|" + reply + ")$", "read": "^" + incoming + "$"}, + {"user": "invemory_staging", "vhost": vhost, "configure": "^(" + incoming + "|" + reply + ")$", "write": "^(amq\\.default|" + incoming + ")$", "read": "^" + reply + "$"}], + "queues": [{"name": name, "vhost": vhost, "durable": True, "auto_delete": False, "arguments": {}} for name in ("notification.send-email", config["replyQueue"])], "exchanges": [], "bindings": []} + + +def secret_manifests(config, credentials, registry): + def secret(name, data, kind="Opaque"): + return obj("Secret", name, type=kind, data={key: base64.b64encode(value.encode()).decode() for key, value in data.items()}) + + data = {key: credentials[key] for key in ("postgresPassword", "runtimePassword", "rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword", "seqApiKey")} + data["ownerConnectionString"] = "Host=postgres;Database=notification_service;Username=notification_owner;Password=" + credentials["postgresPassword"] + data["runtimeConnectionString"] = "Host=postgres;Database=notification_service;Username=notification_runtime;Password=" + credentials["runtimePassword"] + return [secret("notification-private", data), secret("rabbit-definitions", {"definitions.json": json.dumps(rabbit_definitions(config, credentials))}), + secret("registry-pull", {".dockerconfigjson": json.dumps(registry)}, "kubernetes.io/dockerconfigjson")] + + +class Cluster: + def __init__(self, args): + self.command = [args.kubectl, "--kubeconfig=" + args.kubeconfig, "--request-timeout=20s"] + + def run(self, *args, data=None, timeout=120, sensitive=False): + result = subprocess.run(self.command + list(args), input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout) + if result.returncode: + # Error output can contain submitted secret resources or provider details. + detail = "" if sensitive else result.stderr.decode(errors="replace")[:1000] + raise RuntimeError("kubectl operation failed: " + " ".join(args[:3]) + ("\n" + detail if detail else "")) + return result.stdout + + def get(self, kind, name=None, namespace=NAMESPACE): + args = ["get", kind] + if name: + args.append(name) + if namespace: + args += ["-n", namespace] + args += ["-o", "json"] + if name: + args.append("--ignore-not-found") + data = self.run(*args) + if not data.strip(): + if name: + return None + raise RuntimeError("kubectl returned no JSON for resource listing: " + kind) + value = json.loads(data) + if not name and (not isinstance(value, dict) or not isinstance(value.get("items"), list)): + raise RuntimeError("kubectl returned an unexpected resource list: " + kind) + return value + + def apply(self, values, dry_run=False, sensitive=False): + args = ["apply", "-f", "-"] + if dry_run: + args += ["--dry-run=server"] + self.run(*args, data=json.dumps({"apiVersion": "v1", "kind": "List", "items": values}).encode(), sensitive=sensitive) + + def scale(self, component, replicas): + if self.get("deployment", component): + self.run("scale", "deployment/" + component, "-n", NAMESPACE, "--replicas=" + str(replicas)) + if replicas == 0 and any(p["metadata"].get("labels", {}).get("app.kubernetes.io/component") == component for p in self.get("pods")["items"]): + self.run("wait", "--for=delete", "pod", "-n", NAMESPACE, "-l", "app.kubernetes.io/component=" + component, "--timeout=120s", timeout=140) + + def rollout(self, component): + self.run("rollout", "status", "deployment/" + component, "-n", NAMESPACE, "--timeout=240s", timeout=260) + + def history(self): + table = self.run("exec", "deployment/postgres", "-n", NAMESPACE, "--", "psql", "-U", "notification_owner", "-d", "notification_service", "-At", "-c", "SELECT to_regclass('public.\"__EFMigrationsHistory\"')").decode().strip() + if not table: + return [] + return self.run("exec", "deployment/postgres", "-n", NAMESPACE, "--", "psql", "-U", "notification_owner", "-d", "notification_service", "-At", "-c", 'SELECT "MigrationId" FROM "__EFMigrationsHistory" ORDER BY "MigrationId"').decode().splitlines() + + +@contextlib.contextmanager +def lock(directory): + directory = Path(directory) + directory.mkdir(parents=True, exist_ok=True, mode=0o700) + path = directory / "deployment.lock" + try: + fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + except FileExistsError: + raise RuntimeError("deployment lock exists; verify no active deployment before manually removing a stale lock") + try: + os.write(fd, str(os.getpid()).encode()) + os.close(fd) + yield + finally: + path.unlink() + + +def preflight(cluster, config, resources): + node = cluster.get("node", config["node"], namespace=None) + if not node or not any(x["type"] == "Ready" and x["status"] == "True" for x in node["status"]["conditions"]): + raise RuntimeError("target storage node is not Ready") + if any(x["type"] in ("DiskPressure", "MemoryPressure", "PIDPressure") and x["status"] == "True" for x in node["status"]["conditions"]): + raise RuntimeError("target node has resource pressure") + seq_pods = cluster.get("pods", namespace=config["seqNamespace"])["items"] + if not any(all(p["metadata"].get("labels", {}).get(k) == v for k, v in config["seqPodLabels"].items()) for p in seq_pods): + raise RuntimeError("Seq selector matches no existing pod; verify dependency labels") + for item in resources: + kind, name = item["kind"], item["metadata"]["name"] + current = cluster.get(kind, name, item["metadata"].get("namespace")) + if not current: + continue + if current["metadata"].get("labels", {}).get("app.kubernetes.io/managed-by") != MANAGED: + raise RuntimeError("existing resource is not owned by this package: " + kind + "/" + name) + if kind == "PersistentVolume": + old = current["spec"] + for key in ("local", "nodeAffinity", "capacity", "persistentVolumeReclaimPolicy"): + if old.get(key) != item["spec"][key]: + raise RuntimeError("storage changes require a separate migration procedure") + if current["metadata"].get("annotations", {}).get("hrynco.com/storage-uuid") != config["storage"]["uuid"]: + raise RuntimeError("storage UUID change requires a separate migration procedure") + if kind == "Deployment" and name in ("postgres", "rabbitmq"): + if current["spec"]["template"]["spec"]["containers"][0]["image"] != item["spec"]["template"]["spec"]["containers"][0]["image"]: + raise RuntimeError("database/broker image change requires separate backup/upgrade procedure") + + +def host_storage_check(config): + """Run on the storage node, not on the developer workstation.""" + if sys.platform != "linux": + raise RuntimeError("host storage preflight and deployment must run on the Linux storage node") + for component, uid, gid in (("postgres", 999, 999), ("rabbitmq", 100, 101)): + path = config["storage"]["basePath"] + "/" + component + result = subprocess.run(["findmnt", "--json", "--target", path, "--output", "FSTYPE,UUID"], capture_output=True, check=True, timeout=20) + filesystems = json.loads(result.stdout)["filesystems"] + if len(filesystems) != 1 or filesystems[0]["fstype"] != config["storage"]["filesystem"] or filesystems[0]["uuid"] != config["storage"]["uuid"]: + raise RuntimeError("host storage filesystem/UUID does not match reviewed configuration") + info = os.stat(path) + if info.st_uid != uid or info.st_gid != gid or info.st_mode & 0o007: + raise RuntimeError("storage ownership or world permissions differ from provisioned contract") + + +def job(component, image, command=None, environment=None, volumes=None, mounts=None): + container = {"name": component, "image": image, "securityContext": secure(), + "resources": {"requests": {"cpu": "100m", "memory": "128Mi"}, "limits": {"cpu": "500m", "memory": "512Mi"}}, "env": environment or []} + if command: + container["command"] = command + if mounts: + container["volumeMounts"] = mounts + pod = {"restartPolicy": "Never", "automountServiceAccountToken": False, + "imagePullSecrets": [{"name": "registry-pull"}], "containers": [container]} + if volumes: + pod["volumes"] = volumes + return obj("Job", "notification-" + component, {"backoffLimit": 0, "activeDeadlineSeconds": 300, + "template": {"metadata": {"labels": labels(component)}, "spec": pod}}, api="batch/v1") + + +def run_job(cluster, value): + name = value["metadata"]["name"] + cluster.run("delete", "job", name, "-n", NAMESPACE, "--ignore-not-found", "--wait=true", "--timeout=60s") + cluster.apply([value]) + cluster.run("wait", "--for=condition=complete", "job/" + name, "-n", NAMESPACE, "--timeout=300s", timeout=320) + + +def backup(cluster, private_dir): + folder = Path(private_dir) / ("backup-" + datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")) + folder.mkdir(mode=0o700) + dump = cluster.run("exec", "deployment/postgres", "-n", NAMESPACE, "--", "pg_dump", "-U", "notification_owner", "-d", "notification_service", "-Fc", timeout=300, sensitive=True) + if not dump.startswith(b"PGDMP"): + raise RuntimeError("invalid PostgreSQL backup") + fd = os.open(folder / "database.dump", os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as stream: + stream.write(dump) + cluster.run("exec", "-i", "deployment/postgres", "-n", NAMESPACE, "--", "pg_restore", "--list", data=dump, sensitive=True) + write_private(folder / "manifest.json", {"sha256": hashlib.sha256(dump).hexdigest(), "bytes": len(dump), "history": cluster.history(), "scope": "notification_service database only; broker PV retained, no offsite proof"}) + return folder.name + + +def database_ready(deployment): + if not deployment or deployment.get("spec", {}).get("replicas") != 1: + return False + status = deployment.get("status", {}) + return (status.get("readyReplicas", 0) == 1 and status.get("updatedReplicas", 0) == 1 and + status.get("observedGeneration", 0) >= deployment.get("metadata", {}).get("generation", 1)) + + +def recovery_foundation(resources): + # Ownership/storage/image preflight has already passed. Do not apply Web/Worker + # targets or start their writers before history and a verified database backup. + return [value for value in resources if not ( + value["kind"] in ("Deployment", "Service") and value["metadata"]["name"] in ("web", "worker"))] + + +def wait_for_worker_consumer(cluster, config, timeout=60): + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RuntimeError("expected exactly one notification.send-email consumer before timeout") + try: + output = cluster.run("exec", "deployment/rabbitmq", "-n", NAMESPACE, "--", "rabbitmqctl", "-q", + "list_queues", "-p", config["rabbitVirtualHost"], "name", "consumers", + timeout=min(20, remaining), sensitive=True).decode() + except (RuntimeError, subprocess.TimeoutExpired): + raise RuntimeError("could not verify notification.send-email consumer count") from None + rows = [line.split() for line in output.splitlines() if line.split()[:1] == ["notification.send-email"]] + if len(rows) != 1 or len(rows[0]) != 2 or rows[0][1] not in ("0", "1"): + raise RuntimeError("expected exactly one notification.send-email consumer; unexpected broker count") + remaining = deadline - time.monotonic() + if remaining <= 0: + raise RuntimeError("expected exactly one notification.send-email consumer before timeout") + if rows[0][1] == "1": + return + # Process rollout can complete before the RabbitMQ subscription attaches. + # Only zero consumers is a transient state; extra/unknown counts fail closed. + time.sleep(min(2, remaining)) + + +def deploy(cluster, config, artifacts, args): + resources = manifest(config, artifacts) + preflight(cluster, config, resources) + host_storage_check(config) + exists = cluster.get("namespace", NAMESPACE, namespace=None) is not None + if not exists: + raise RuntimeError("run bootstrap first, then inspect node capacity and server dry-run before deploy") + if not cluster.get("secret", "notification-private") or not cluster.get("secret", "registry-pull"): + raise RuntimeError("private credentials and registry pull Secret required") + with lock(args.private_dir): + # Never let an ordinary redeploy bypass migrations/schema validation. + database = cluster.get("deployment", "postgres") + current_worker = cluster.get("deployment", "worker") + worker_replicas = current_worker["spec"].get("replicas", 0) if current_worker else 0 + if args.enable_worker and not args.sandbox_verified: + raise RuntimeError("inspect sandbox channel/template and broker backlog before enabling the Worker") + if not database_ready(database): + if not args.migrate: + raise RuntimeError("PostgreSQL is not ready; interrupted install/recovery requires --migrate") + cluster.scale("worker", 0) + cluster.scale("web", 0) + cluster.apply(recovery_foundation(resources)) + for component in ("postgres", "rabbitmq", "mailpit"): + cluster.rollout(component) + # Deployment existence does not establish a running database or empty disk. + # After recovery, query the actual retained DB rather than assume [] history. + before = cluster.history() + if before != MIGRATIONS and not args.migrate: + raise RuntimeError("schema is not the reviewed target; --migrate required") + if before != MIGRATIONS[:len(before)]: + raise RuntimeError("database has unknown or non-prefix migration history") + if args.migrate: + cluster.scale("worker", 0) + cluster.scale("web", 0) + # Always back up the actual DB, including an empty initialized first DB. + # A removed Deployment or interrupted first install may retain real data. + backup(cluster, args.private_dir) + # Preserve runtime replicas during normal reconciliation. First install stays stopped. + for value in resources: + if value["kind"] == "Deployment" and value["metadata"]["name"] in ("web", "worker"): + value["spec"]["replicas"] = 0 if args.migrate else (1 if value["metadata"]["name"] == "web" else worker_replicas) + cluster.apply(resources) + for component in ("postgres", "rabbitmq", "mailpit"): + cluster.rollout(component) + if args.migrate: + run_job(cluster, job("migrator", artifacts["images"]["migrator"]["image"], environment=[ + env("DOTNET_ENVIRONMENT", "Production"), env("App__ConnectionString", secret="notification-private", key="ownerConnectionString")])) + if cluster.history() != MIGRATIONS: + raise RuntimeError("migrator completed but schema history is not the reviewed target") + run_job(cluster, job("grants", artifacts["images"]["postgres"]["image"], + ["psql", "-h", "postgres", "-U", "notification_owner", "-d", "notification_service", "-v", "ON_ERROR_STOP=1", "-q", "-f", "/sql/grants.sql"], + [env("PGPASSWORD", secret="notification-private", key="postgresPassword"), env("RUNTIME_PASSWORD", secret="notification-private", key="runtimePassword")], + [{"name": "sql", "configMap": {"name": "runtime-grants"}}], [{"name": "sql", "mountPath": "/sql", "readOnly": True}])) + run_job(cluster, job("sandbox", artifacts["images"]["postgres"]["image"], + ["psql", "-h", "postgres", "-U", "notification_owner", "-d", "notification_service", "-v", "ON_ERROR_STOP=1", "-q", "-f", "/sql/sandbox.sql"], + [env("PGPASSWORD", secret="notification-private", key="postgresPassword")], + [{"name": "sql", "configMap": {"name": "sandbox-seed"}}], [{"name": "sql", "mountPath": "/sql", "readOnly": True}])) + cluster.scale("web", 1) + cluster.rollout("web") + # Worker startup is separate: inspect/seed sandbox config before consuming anything. + if args.enable_worker: + safe = cluster.run("exec", "deployment/postgres", "-n", NAMESPACE, "--", "psql", "-U", "notification_owner", "-d", "notification_service", "-At", "-c", SANDBOX_CHECK).decode().strip() + if safe != "t": + raise RuntimeError("sandbox channel/template configuration does not match the safe seed") + queues = cluster.run("exec", "deployment/rabbitmq", "-n", NAMESPACE, "--", "rabbitmqctl", "-q", "list_queues", "-p", config["rabbitVirtualHost"], "name", "messages_ready", "messages_unacknowledged").decode() + if not any(line.split() == ["notification.send-email", "0", "0"] for line in queues.splitlines()): + raise RuntimeError("incoming queue is not empty; inspect backlog before enabling the Worker") + cluster.scale("worker", 1) + # A preserved running Worker is also a required gate during ordinary deploy. + if args.enable_worker or (not args.migrate and worker_replicas > 0): + try: + cluster.rollout("worker") + wait_for_worker_consumer(cluster, config) + except Exception: + cluster.scale("worker", 0) + raise + print("deployment complete; SMTP is sandbox-only, Invemory publisher cutover is separate") + + +def main(argv=None): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("command", choices=("render", "check", "bootstrap", "deploy", "backup")) + parser.add_argument("--config", required=True) + parser.add_argument("--artifacts", required=True) + parser.add_argument("--kubectl", default="/snap/microk8s/current/kubectl") + parser.add_argument("--kubeconfig", default="/home/devops/.kube/cluster-admin.conf") + parser.add_argument("--private-dir", default="/home/devops/shared-notification-private") + parser.add_argument("--output") + parser.add_argument("--seq-key-file") + parser.add_argument("--registry-auth-file") + parser.add_argument("--migrate", action="store_true") + parser.add_argument("--enable-worker", action="store_true") + parser.add_argument("--sandbox-verified", action="store_true") + args = parser.parse_args(argv) + config, artifacts = read_json(args.config), read_json(args.artifacts) + resources = manifest(config, artifacts) + if args.command == "render": + if not args.output: + raise ValueError("render requires --output") + Path(args.output).write_text(json.dumps({"apiVersion": "v1", "kind": "List", "items": resources}, indent=2) + "\n", encoding="utf-8") + print("rendered reviewed resource package") + return + cluster = Cluster(args) + if args.command == "bootstrap": + if sys.platform != "linux": + raise RuntimeError("private secret bootstrap must run on the Linux deployment host") + if not args.seq_key_file or not args.registry_auth_file: + raise ValueError("bootstrap requires private dedicated Seq ingest key and registry auth files") + path = Path(args.private_dir) / "credentials.json" + current_secret = cluster.get("secret", "notification-private") + if path.exists(): + credentials = read_json(path) + else: + if current_secret: + raise RuntimeError("private credential file missing while live Secret exists; recover original private file; do not regenerate passwords") + credentials = {key: secrets.token_hex(32) for key in ("postgresPassword", "runtimePassword", "rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword")} + for key in ("rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword"): + credentials[key + "Salt"] = base64.b64encode(secrets.token_bytes(4)).decode() + credentials["seqApiKey"] = Path(args.seq_key_file).read_text(encoding="utf-8").strip() + if not credentials["seqApiKey"]: + raise ValueError("dedicated Seq key is empty") + write_private(path, credentials) + if credentials["seqApiKey"] != Path(args.seq_key_file).read_text(encoding="utf-8").strip(): + raise RuntimeError("dedicated Seq key differs from preserved credentials; explicit rotation procedure required") + registry = read_json(args.registry_auth_file) + if set(registry.get("auths", {})) != {"registry.hrynco.com"}: + raise ValueError("registry auth must contain registry.hrynco.com only") + namespace = cluster.get("namespace", NAMESPACE, namespace=None) + if namespace and namespace["metadata"].get("labels", {}).get("app.kubernetes.io/managed-by") != MANAGED: + raise RuntimeError("refusing to adopt existing namespace") + cluster.apply([resources[0]]) + values = secret_manifests(config, credentials, registry) + for value in values: + current = cluster.get("secret", value["metadata"]["name"]) + if current and current["metadata"].get("labels", {}).get("app.kubernetes.io/managed-by") != MANAGED: + raise RuntimeError("refusing to adopt existing Secret") + if current and value["metadata"]["name"] != "registry-pull" and current.get("data") != value["data"]: + raise RuntimeError("preserved credentials or broker definitions differ from live Secret; explicit rotation/routing procedure required") + cluster.apply(values, sensitive=True) + print("private credentials preserved and Secrets reconciled; values not printed") + elif args.command == "check": + preflight(cluster, config, resources) + host_storage_check(config) + if not cluster.get("namespace", NAMESPACE, namespace=None): + raise RuntimeError("bootstrap namespace/Secrets before server-side dry-run") + cluster.apply(resources, dry_run=True) + print("ownership, host storage UUID, pinned artifacts and server-side dry-run passed; measured node capacity remains mandatory") + elif args.command == "backup": + with lock(args.private_dir): + if any((cluster.get("deployment", x) or {}).get("spec", {}).get("replicas", 0) for x in ("web", "worker")): + raise RuntimeError("stop Web/Worker writers before backup") + print("verified local backup: " + backup(cluster, args.private_dir)) + else: + deploy(cluster, config, artifacts, args) + + +if __name__ == "__main__": + try: + main() + except (ValueError, RuntimeError, KeyError, OSError, subprocess.TimeoutExpired, subprocess.CalledProcessError) as error: + print("STOP: " + str(error), file=sys.stderr) + sys.exit(1) diff --git a/docker/kubernetes/shared-notification/prepare-seq-key.py b/docker/kubernetes/shared-notification/prepare-seq-key.py new file mode 100644 index 0000000..8bc2629 --- /dev/null +++ b/docker/kubernetes/shared-notification/prepare-seq-key.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +"""Provision only the dedicated Notification ingest key; never print credentials.""" + +import contextlib +import copy +from datetime import datetime, timezone +import http.cookiejar +import json +import os +from pathlib import Path +import re +import secrets +import stat +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + +BASE = "http://192.168.2.44:30534/" +CREDENTIALS = Path("/home/devops/seq-first/credentials.json") +TOKEN_FILE = Path("/home/devops/notification-private/seq-key.txt") +TITLE = "HrynCo Notification Service" +PROPERTIES = [{"Name": "Environment", "Value": "SharedNotification"}, + {"Name": "Service", "Value": "HrynCo.NotificationService"}] +MARKER = "NotificationIngestKeyId" +KEY_LIST = "api/apikeys?shared=true" + + +def allowed_path(method, path): + methods = {"api/users/login": "POST", KEY_LIST: "GET", "api/apikeys/template": "GET", + "api/apikeys/": "POST", "api/events/raw": "POST"} + return methods.get(path) == method or (method == "GET" and re.fullmatch( + r"api/events\?count=2&filter=ProvisioningCheckId%20%3D%20%27[a-f0-9]{32}%27", path)) or ( + method == "PUT" and re.fullmatch(r"api/apikeys/apikey-[a-z0-9-]+", path)) + + +def require(condition, message): + if not condition: + raise RuntimeError(message) + + +class NoRedirects(urllib.request.HTTPRedirectHandler): + def redirect_request(self, request, fp, code, message, headers, new_url): + raise RuntimeError("unexpected Seq redirect denied") + + +class Client: + def __init__(self): + self.session = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirects(), + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())) + self.csrf = None + + def request(self, method, path, body=None, key=None): + require(allowed_path(method, path), "unexpected Seq API path") + if method == "PUT": + verify_key(body) + require(path == "api/apikeys/" + body["Id"], "dedicated Seq update path mismatch") + headers = {} + if isinstance(body, dict): + body = json.dumps(body).encode() + headers["Content-Type"] = "application/json" + if key: + require(path == "api/events/raw", "ingestion token used outside ingestion") + headers["X-Seq-ApiKey"] = key + headers["Content-Type"] = "application/vnd.serilog.clef" + opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirects()) + else: + opener = self.session + if self.csrf and method != "GET": + headers["X-Seq-CsrfToken"] = self.csrf + request = urllib.request.Request(BASE + path, data=body, headers=headers, method=method) + try: + with opener.open(request, timeout=15) as response: + require(response.geturl() == BASE + path, "unexpected Seq response endpoint") + raw = response.read(1024 * 1024 + 1) + require(len(raw) <= 1024 * 1024, "unexpected Seq response size") + require(response.status in (200, 201, 202, 204), "Seq request failed") + return json.loads(raw) if raw and "json" in response.headers.get("Content-Type", "") else None + except urllib.error.HTTPError as error: + code = error.code + error.close() + raise RuntimeError("Seq request denied (HTTP " + str(code) + ")") from None + except (urllib.error.URLError, TimeoutError): + raise RuntimeError("trusted Seq endpoint unavailable") from None + + def login(self, credentials): + response = self.request("POST", "api/users/login", { + "Username": credentials["adminUsername"], "Password": credentials["adminPassword"]}) + require(response["Username"] == credentials["adminUsername"] and + isinstance(response.get("CsrfToken"), str) and response["CsrfToken"], "Seq login identity/CSRF mismatch") + self.csrf = response["CsrfToken"] + + +def private_directory(path): + require(not path.is_symlink(), "private directory must not be a symlink") + path.mkdir(mode=0o700, parents=True, exist_ok=True) + require(path.is_dir(), "private directory is not a directory") + if os.name == "posix": + info = path.stat() + require(info.st_uid == os.getuid() and stat.S_IMODE(info.st_mode) == 0o700, + "private directory ownership/mode must match devops/0700") + + +def private_file(path): + require(not path.is_symlink() and path.is_file(), "private file missing or unexpected type") + if os.name == "posix": + info = path.stat() + require(info.st_uid == os.getuid() and stat.S_IMODE(info.st_mode) == 0o600, + "private file ownership/mode must match devops/0600") + + +@contextlib.contextmanager +def key_lock(directory): + path = directory / ".seq-key.lock" + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError: + raise RuntimeError("Seq key lock exists; inspect active run before removing stale lock") from None + try: + os.close(fd) + yield + finally: + path.unlink() + + +def verify_key(value, allow_unmarked=False): + require(isinstance(value, dict), "unexpected dedicated Seq key response") + require(value.get("Title") == TITLE and isinstance(value.get("Id"), str) and + re.fullmatch(r"apikey-[a-z0-9-]+", value["Id"]), "dedicated Seq key identity mismatch") + require(value.get("AssignedPermissions") == ["Ingest"], "dedicated Seq key must grant only Ingest") + properties = value.get("InputSettings", {}).get("AppliedProperties") + expected = PROPERTIES + [{"Name": MARKER, "Value": value["Id"]}] + unmarked = allow_unmarked and isinstance(properties, list) and len(properties) == len(PROPERTIES) and ( + sorted(properties, key=lambda x: x["Name"]) == sorted(PROPERTIES, key=lambda x: x["Name"])) + require(isinstance(properties, list) and (unmarked or ( + len(properties) == len(expected) and sorted(properties, key=lambda x: x["Name"]) == sorted(expected, key=lambda x: x["Name"]))), + "dedicated Seq key stamped properties mismatch") + + +def dedicated_keys(client): + values = client.request("GET", KEY_LIST) + require(isinstance(values, list), "unexpected Seq key list") + matches = [value for value in values if value.get("Title") == TITLE] + require(len(matches) <= 1, "ambiguous dedicated Seq keys; no key changed") + return matches + + +def marker_unique(values, key_id): + for value in values: + if value.get("Id") == key_id: + continue + properties = value.get("InputSettings", {}).get("AppliedProperties") or [] + require(isinstance(properties, list) and all(isinstance(p, dict) for p in properties), "unexpected Seq marker property schema") + require(not any(p.get("Name") == MARKER and p.get("Value") == key_id for p in properties), + "dedicated Seq identity marker collision; no foreign key changed") + + +def ensure_identity_marker(client, key_id): + values = client.request("GET", KEY_LIST) + require(isinstance(values, list), "unexpected Seq key list") + targets = [v for v in values if v.get("Id") == key_id and v.get("Title") == TITLE] + require(len(targets) == 1, "dedicated Seq key changed before guarded update") + target = targets[0] + verify_key(target, allow_unmarked=True) + marker_unique(values, key_id) + expected_settings = copy.deepcopy(target["InputSettings"]) + if not any(p["Name"] == MARKER for p in expected_settings["AppliedProperties"]): + expected_settings["AppliedProperties"].append({"Name": MARKER, "Value": key_id}) + updated = copy.deepcopy(target) + updated["InputSettings"] = expected_settings + # Seq's documented UpdateAsync cannot change the token. Preserve every + # existing entity field; only this dedicated key's one stamp is added. + client.request("PUT", "api/apikeys/" + key_id, updated) + readback = client.request("GET", KEY_LIST) + require(isinstance(readback, list), "unexpected Seq key readback") + marker_unique(readback, key_id) + targets = [v for v in readback if v.get("Id") == key_id and v.get("Title") == TITLE] + require(len(targets) == 1, "dedicated Seq key readback missing") + result = targets[0] + verify_key(result) + require(result.get("InputSettings") == expected_settings and + all(result.get(field) == target.get(field) for field in ("AssignedPermissions", "OwnerId", "IsDefault", "Token", "TokenPrefix")), + "dedicated Seq update changed settings beyond identity marker") + return result + + +def verify_canary(client, check_id, expected_key): + query = "api/events?count=2&filter=" + urllib.parse.quote("ProvisioningCheckId = '" + check_id + "'") + events = [] + for attempt in range(20): + events = client.request("GET", query) + require(isinstance(events, list), "unexpected Seq canary readback") + if events: + break + if attempt < 19: + time.sleep(1) + require(len(events) == 1 and isinstance(events[0], dict), "Seq canary readback must contain exactly one event") + event = events[0] + # Seq's official EventEntity exposes Properties as EventPropertyPart{Name,Value}. + parts = event.get("Properties") + require(isinstance(parts, list) and all(isinstance(p, dict) and "Name" in p and "Value" in p for p in parts), + "unexpected Seq canary property schema") + properties = {p["Name"]: p["Value"] for p in parts} + require(len(properties) == len(parts), "ambiguous Seq canary properties") + require(properties.get("ProvisioningCheckId") == check_id, "Seq canary correlation mismatch") + # The canary payload omits this server-stamped, globally unique key marker. + # Installed Seq exposes neither the original token nor ApiKeyId on EventEntity. + require(properties.get(MARKER) == expected_key["Id"], "Seq canary token belongs to a different or unverified key") + actual_key = event.get("ApiKeyId", properties.get("ApiKeyId")) + if actual_key is not None: + require(actual_key == expected_key["Id"], "Seq canary provenance conflicts with identity marker") + require(properties.get("Environment") == "SharedNotification" and + properties.get("Service") == "HrynCo.NotificationService", "Seq canary stamped identity mismatch") + + +def provision(client, token_file=TOKEN_FILE): + private_directory(token_file.parent) + with key_lock(token_file.parent): + existing = dedicated_keys(client) + if existing: + verify_key(existing[0], allow_unmarked=True) + require(token_file.exists(), "dedicated remote key exists but local token is missing; recover original file; no rotation performed") + if token_file.exists(): + private_file(token_file) + token = token_file.read_text(encoding="utf-8").strip() + require(re.fullmatch(r"[a-f0-9]{64}", token), "private token format mismatch; no rotation performed") + else: + token = secrets.token_hex(32) + # Persist BEFORE POST: a timed-out create can be recovered without rotation. + fd = os.open(token_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + stream.write(token + "\n") + stream.flush() + os.fsync(stream.fileno()) + if not existing: + template = client.request("GET", "api/apikeys/template") + require(isinstance(template, dict) and isinstance(template.get("InputSettings"), dict), "unexpected Seq key template") + template.update({"Title": TITLE, "Token": token, "AssignedPermissions": ["Ingest"]}) + template["InputSettings"]["AppliedProperties"] = PROPERTIES + created = client.request("POST", "api/apikeys/", template) + verify_key(created, allow_unmarked=True) + readback = dedicated_keys(client) + require(len(readback) == 1, "dedicated Seq key readback missing") + dedicated = ensure_identity_marker(client, readback[0]["Id"]) + # A valid foreign token must not pass. The marker is applied by Seq only, + # never supplied by this canary's body, and is unique across all keys. + canary = {"@t": datetime.now(timezone.utc).isoformat(), + "@mt": "HrynCo Notification Service ingest key verification", + "Application": "hrynco-notification-service-provisioning", + "ProvisioningCheckId": secrets.token_hex(16)} + client.request("POST", "api/events/raw", json.dumps(canary).encode() + b"\n", key=token) + verify_canary(client, canary["ProvisioningCheckId"], dedicated) + return dedicated["Id"] + + +def main(): + require(sys.platform == "linux" and os.getuid() != 0, "run as devops on mini-pc-2") + import pwd + require(pwd.getpwuid(os.getuid()).pw_name == "devops", "run as devops on mini-pc-2") + private_file(CREDENTIALS) + credentials = json.loads(CREDENTIALS.read_text(encoding="utf-8")) + client = Client() + client.login(credentials) + provision(client) + print("Dedicated Notification Seq key/marker verified; private token preserved; other keys and retention unchanged.") + + +if __name__ == "__main__": + try: + main() + except (RuntimeError, OSError, ValueError, KeyError, TypeError) as error: + # RuntimeError messages are fixed gates; never API responses or file contents. + message = str(error) if isinstance(error, RuntimeError) else type(error).__name__ + print("STOP: " + message, file=sys.stderr) + sys.exit(1) diff --git a/docker/kubernetes/shared-notification/prepare-storage.py b/docker/kubernetes/shared-notification/prepare-storage.py new file mode 100755 index 0000000..b1ea111 --- /dev/null +++ b/docker/kubernetes/shared-notification/prepare-storage.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +"""Owner-run provisioning/verification of Notification Service storage. No deletion.""" + +import json +import os +from pathlib import Path +import stat +import subprocess +import sys + +BASE = Path("/mnt/shared/notification-service") +UUID = "c2c31640-a247-4988-bc7b-9ee9789464d1" +DIRECTORIES = (("postgres", 999, 999), ("rabbitmq", 100, 101)) + + +def main(): + if sys.platform != "linux" or os.geteuid() != 0: + raise RuntimeError("run this script as root on mini-pc-2") + for location in (Path("/mnt/files"), Path("/mnt/shared")): + if not location.is_dir() or location.is_symlink(): + raise RuntimeError("expected real storage mount directory is missing") + result = subprocess.run(["findmnt", "--json", "--target", str(location), "--output", "TARGET,FSTYPE,UUID"], + check=True, capture_output=True, timeout=20) + filesystems = json.loads(result.stdout)["filesystems"] + if len(filesystems) != 1 or filesystems[0]["target"] != str(location) or filesystems[0]["fstype"] != "xfs" or filesystems[0]["uuid"] != UUID: + raise RuntimeError("storage mount/UUID/XFS verification failed; no directories created") + if BASE.is_symlink(): + raise RuntimeError("notification-service path must not be a symlink") + if BASE.exists(): + info = BASE.stat() + if not BASE.is_dir() or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (0, 0, 0o755): + raise RuntimeError("existing parent path has unexpected type/ownership/permissions") + if set(child.name for child in BASE.iterdir()) - {name for name, _, _ in DIRECTORIES}: + raise RuntimeError("unexpected contents in notification-service parent") + # Inspect both directories before creating anything. Never adopt arbitrary data. + for name, uid, gid in DIRECTORIES: + path = BASE / name + if path.is_symlink(): + raise RuntimeError("service data path must not be a symlink") + if not path.exists(): + continue + info = path.stat() + if not path.is_dir() or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (uid, gid, 0o700): + raise RuntimeError("existing service directory has unexpected ownership/permissions") + marker = path / ".invemory-storage.json" + if marker.is_symlink(): + raise RuntimeError("marker must not be a symlink") + if marker.exists(): + marker_info = marker.stat() + if (marker_info.st_uid, marker_info.st_gid, stat.S_IMODE(marker_info.st_mode)) != (0, 0, 0o644): + raise RuntimeError("existing marker has unexpected ownership/permissions") + if json.loads(marker.read_text(encoding="utf-8")) != {"uuid": UUID, "role": "shared-notification/" + name}: + raise RuntimeError("existing marker mismatch; never overwritten") + elif any(path.iterdir()): + raise RuntimeError("unmarked service directory is not empty; never adopted") + os.umask(0o077) + if not BASE.exists(): + BASE.mkdir(mode=0o755) + os.chown(BASE, 0, 0) + os.chmod(BASE, 0o755) + for name, uid, gid in DIRECTORIES: + path = BASE / name + if not path.exists(): + path.mkdir(mode=0o700) + os.chown(path, uid, gid) + marker = path / ".invemory-storage.json" + if not marker.exists(): + fd = os.open(marker, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644) + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump({"uuid": UUID, "role": "shared-notification/" + name}, stream) + stream.write("\n") + os.chown(marker, 0, 0) + os.chmod(marker, 0o644) + info = path.stat() + marker_info = marker.stat() + if (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (uid, gid, 0o700): + raise RuntimeError("directory ownership verification failed") + if (marker_info.st_uid, marker_info.st_gid, stat.S_IMODE(marker_info.st_mode)) != (0, 0, 0o644): + raise RuntimeError("marker ownership verification failed") + print("Provisioned two Notification Service directories; XFS UUID and permissions verified.") + + +if __name__ == "__main__": + try: + main() + except (RuntimeError, OSError, ValueError, subprocess.SubprocessError) as error: + print("STOP: " + str(error), file=sys.stderr) + sys.exit(1) diff --git a/docker/kubernetes/shared-notification/tests/test_notification.py b/docker/kubernetes/shared-notification/tests/test_notification.py new file mode 100644 index 0000000..de7aef4 --- /dev/null +++ b/docker/kubernetes/shared-notification/tests/test_notification.py @@ -0,0 +1,446 @@ +import argparse +import base64 +import copy +import importlib.util +import json +from pathlib import Path +import re +import tempfile +import unittest +from unittest.mock import patch + + +PACKAGE = Path(__file__).resolve().parents[1] + + +def load_module(name, filename): + spec = importlib.util.spec_from_file_location(name, PACKAGE / filename) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +ns = load_module("notification", "notification.py") +storage = load_module("prepare_storage", "prepare-storage.py") + + +def fixture(): + config = ns.read_json(PACKAGE / "config.example.json") + artifacts = ns.read_json(PACKAGE / "artifacts.example.json") + artifacts["sourceCommit"] = "a" * 40 + artifacts["sourceArchiveSha256"] = "b" * 64 + for key, value in artifacts["images"].items(): + value["image"] = "registry.hrynco.com/test/" + key + "@sha256:" + "c" * 64 + if key in ("web", "worker", "migrator"): + value["sourceArchiveSha256"] = artifacts["sourceArchiveSha256"] + return config, artifacts + + +class FakeCluster: + def __init__(self, history=None, worker=0, worker_error=False, consumer_count=1, + postgres_ready=True, postgres_exists=True, storage_error=False): + self.schema = list(ns.MIGRATIONS if history is None else history) + self.replicas = {"web": 1, "worker": worker, "postgres": 1} + if not postgres_exists: + self.replicas.pop("postgres") + elif not postgres_ready: + self.replicas["postgres"] = 0 + self.events = [] + self.worker_error = worker_error + self.consumer_counts = list(consumer_count) if isinstance(consumer_count, list) else [consumer_count] + self.postgres_ready = postgres_ready + self.storage_error = storage_error + self.applications = [] + + def get(self, kind, name=None, namespace=None): + if kind in ("namespace", "secret"): + return {"metadata": {"labels": ns.labels()}} + if kind == "deployment" and name in self.replicas: + value = {"metadata": {"generation": 1}, "spec": {"replicas": self.replicas[name]}} + if name == "postgres": + value["status"] = {"readyReplicas": int(self.postgres_ready), "updatedReplicas": int(self.postgres_ready), "observedGeneration": 1} + return value + return None + + def history(self): + if not self.postgres_ready or self.replicas.get("postgres", 0) != 1: + raise RuntimeError("cannot query stopped database") + self.events.append(("history",)) + return list(self.schema) + + def scale(self, component, replicas): + self.replicas[component] = replicas + self.events.append(("scale", component, replicas)) + + def apply(self, resources): + self.events.append(("apply",)) + self.applications.append(copy.deepcopy(resources)) + for item in resources: + if item["kind"] == "Deployment": + self.replicas[item["metadata"]["name"]] = item["spec"]["replicas"] + + def rollout(self, component): + self.events.append(("rollout", component)) + if component == "postgres": + if self.storage_error: + raise RuntimeError("postgres rollout failed") + self.postgres_ready = True + if component == "worker" and self.worker_error: + raise RuntimeError("worker rollout failed") + + def run(self, *args, **kwargs): + if "psql" in args: + return b"t\n" + if "messages_ready" in args: + return b"notification.send-email\t0\t0\n" + self.events.append(("broker-consumer-check",)) + count = self.consumer_counts.pop(0) if len(self.consumer_counts) > 1 else self.consumer_counts[0] + return ("notification.send-email\t" + str(count) + "\n").encode() + + +class ContractTests(unittest.TestCase): + def test_example_requires_real_pinned_artifacts(self): + with self.assertRaises(ValueError): + ns.manifest(ns.read_json(PACKAGE / "config.example.json"), ns.read_json(PACKAGE / "artifacts.example.json")) + + def test_rejects_mutable_image_and_different_migrator_source(self): + config, artifacts = fixture() + artifacts["images"]["worker"]["image"] = "registry.hrynco.com/test:latest" + with self.assertRaises(ValueError): + ns.validate(config, artifacts) + config, artifacts = fixture() + artifacts["images"]["migrator"]["sourceArchiveSha256"] = "d" * 64 + with self.assertRaises(ValueError): + ns.validate(config, artifacts) + + def test_retained_prebound_storage_and_nonroot_exact_guard(self): + config, artifacts = fixture() + resources = ns.manifest(config, artifacts) + pvs = [x for x in resources if x["kind"] == "PersistentVolume"] + self.assertEqual(len(pvs), 2) + for pv in pvs: + self.assertEqual(pv["spec"]["persistentVolumeReclaimPolicy"], "Retain") + self.assertEqual(pv["spec"]["claimRef"]["namespace"], ns.NAMESPACE) + self.assertTrue(pv["spec"]["local"]["path"].startswith("/mnt/shared/notification-service/")) + for resource in resources: + if resource["kind"] != "Deployment": + continue + pod = resource["spec"]["template"]["spec"] + self.assertFalse(pod["automountServiceAccountToken"]) + self.assertTrue(pod["containers"][0]["securityContext"]["runAsNonRoot"]) + if resource["metadata"]["name"] in ("postgres", "rabbitmq"): + self.assertNotIn("fsGroup", pod.get("securityContext", {})) + guard = pod["initContainers"][0] + values = {x["name"]: x["value"] for x in guard["env"]} + self.assertEqual(values["EXPECTED_UUID"], config["storage"]["uuid"]) + self.assertEqual(values["EXPECTED_ROLE"], ns.NAMESPACE + "/" + resource["metadata"]["name"]) + if resource["metadata"]["name"] == "postgres": + self.assertFalse(any(m["mountPath"] == "/var/run/postgresql" for m in pod["containers"][0]["volumeMounts"])) + + def test_rabbit_probes_allow_observed_slow_diagnostics_without_rapid_restarts(self): + config, artifacts = fixture() + broker = next(x for x in ns.manifest(config, artifacts) + if x["kind"] == "Deployment" and x["metadata"]["name"] == "rabbitmq") + container = broker["spec"]["template"]["spec"]["containers"][0] + for name in ("startupProbe", "readinessProbe", "livenessProbe"): + probe = container[name] + # The constrained live node measured seven seconds for a healthy CLI call. + self.assertGreaterEqual(probe["timeoutSeconds"], 15) + self.assertGreater(probe["periodSeconds"], probe["timeoutSeconds"]) + self.assertEqual(probe["exec"]["command"][:2], ["rabbitmq-diagnostics", "-q"]) + startup = container["startupProbe"] + self.assertEqual(startup["periodSeconds"] * startup["failureThreshold"], 180) + self.assertEqual(container["readinessProbe"]["exec"]["command"][-1], "check_running") + self.assertGreaterEqual(container["livenessProbe"]["periodSeconds"], 30) + + def test_no_public_ports_or_internet_allowance(self): + config, artifacts = fixture() + resources = ns.manifest(config, artifacts) + for item in resources: + self.assertNotEqual(item["kind"], "Ingress") + if item["kind"] == "Service": + self.assertNotIn("type", item["spec"]) + if item["kind"] == "NetworkPolicy": + for rule in item["spec"].get("egress", []): + for target in rule.get("to", []): + self.assertNotIn("ipBlock", target) + self.assertIn("namespaceSelector", target) + self.assertIn("podSelector", target) + worker = next(x for x in resources if x["kind"] == "Deployment" and x["metadata"]["name"] == "worker") + self.assertEqual(worker["spec"]["replicas"], 0) + + def test_broker_hashes_and_scope_are_deterministic(self): + config, _ = fixture() + credentials = {} + for key in ("rabbitAdminPassword", "rabbitWorkerPassword", "rabbitPublisherPassword"): + credentials[key] = "secret-test-value" + credentials[key + "Salt"] = base64.b64encode(b"salt").decode() + first = ns.rabbit_definitions(config, credentials) + self.assertEqual(first, ns.rabbit_definitions(config, credentials)) + self.assertNotIn("secret-test-value", json.dumps(first)) + worker = next(x for x in first["permissions"] if x["user"] == "notification_worker") + self.assertEqual(worker["read"], r"^notification\.send\-email$") + self.assertNotIn(".*", worker["write"]) + for user in ("notification_worker", "invemory_staging"): + permissions = next(x for x in first["permissions"] if x["user"] == user) + self.assertIsNotNone(re.fullmatch(permissions["write"], "amq.default")) + self.assertIsNone(re.fullmatch(permissions["write"], "unreviewed.exchange")) + + def test_secrets_not_present_in_rendered_resources(self): + config, artifacts = fixture() + self.assertFalse(any(x["kind"] == "Secret" for x in ns.manifest(config, artifacts))) + + def test_storage_provision_rejects_wrong_disk_before_creating(self): + with tempfile.TemporaryDirectory() as folder: + target = Path(folder) / "notification-service" + result = argparse.Namespace(stdout=json.dumps({"filesystems": [{"target": "/mnt/files", "fstype": "ext4", "uuid": "wrong"}]}).encode()) + with patch.object(storage.sys, "platform", "linux"), patch.object(storage.os, "geteuid", return_value=0, create=True), patch.object(storage, "BASE", target), patch.object(storage.Path, "is_dir", return_value=True), patch.object(storage.Path, "is_symlink", return_value=False), patch.object(storage.subprocess, "run", return_value=result): + with self.assertRaises(RuntimeError): + storage.main() + self.assertFalse(target.exists()) + + +class ClusterQueries(unittest.TestCase): + def cluster(self): + return ns.Cluster(argparse.Namespace(kubectl="kubectl", kubeconfig="test-config")) + + def test_empty_pod_listing_uses_json_list_without_ignore_not_found(self): + cluster = self.cluster() + pod_list = {"apiVersion": "v1", "kind": "PodList", "items": []} + with patch.object(cluster, "run", return_value=json.dumps(pod_list).encode()) as run: + self.assertEqual(cluster.get("pods"), pod_list) + self.assertEqual(run.call_args.args, ("get", "pods", "-n", ns.NAMESPACE, "-o", "json")) + + def test_missing_named_resource_retains_ignore_not_found_semantics(self): + cluster = self.cluster() + with patch.object(cluster, "run", return_value=b"") as run: + self.assertIsNone(cluster.get("deployment", "worker")) + self.assertIn("--ignore-not-found", run.call_args.args) + + def test_empty_non_json_listing_fails_with_explicit_gate(self): + cluster = self.cluster() + with patch.object(cluster, "run", return_value=b""): + with self.assertRaisesRegex(RuntimeError, "no JSON for resource listing"): + cluster.get("pods") + + def test_stopping_worker_with_no_pods_succeeds_without_wait(self): + cluster = self.cluster() + calls = [] + + def run(*args, **kwargs): + calls.append(args) + if args[:3] == ("get", "deployment", "worker"): + return json.dumps({"spec": {"replicas": 0}}).encode() + if args[:2] == ("get", "pods"): + return json.dumps({"apiVersion": "v1", "kind": "PodList", "items": []}).encode() + if args[:2] == ("scale", "deployment/worker"): + return b"deployment.apps/worker scaled\n" + raise AssertionError("unexpected operation during empty-pod scale") + + with patch.object(cluster, "run", side_effect=run): + cluster.scale("worker", 0) + self.assertTrue(any(call[0] == "scale" for call in calls)) + self.assertFalse(any(call[0] == "wait" for call in calls)) + self.assertNotIn("--ignore-not-found", next(call for call in calls if call[:2] == ("get", "pods"))) + + +class DeploymentGates(unittest.TestCase): + def execute(self, cluster, folder, migrate=False, enable=False, verified=False, backup_error=False, job_error=False): + config, artifacts = fixture() + args = argparse.Namespace(private_dir=folder, migrate=migrate, enable_worker=enable, sandbox_verified=verified) + + def save_backup(client, private_dir): + client.events.append(("backup",)) + if backup_error: + raise RuntimeError("backup failed") + + def run_job(client, value): + client.events.append(("job", value["metadata"]["name"])) + if job_error: + raise RuntimeError("migration failed") + client.schema = list(ns.MIGRATIONS) + + elapsed = [0] + + def sleep(seconds): + elapsed[0] += seconds + + with patch.object(ns, "preflight"), patch.object(ns, "host_storage_check"), patch.object(ns, "backup", side_effect=save_backup), patch.object(ns, "run_job", side_effect=run_job), patch.object(ns.time, "monotonic", side_effect=lambda: elapsed[0]), patch.object(ns.time, "sleep", side_effect=sleep): + ns.deploy(cluster, config, artifacts, args) + + def test_ordinary_repeat_preserves_stopped_worker(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=0) + self.execute(cluster, folder) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertNotIn(("backup",), cluster.events) + self.assertNotIn(("rollout", "worker"), cluster.events) + + def test_ordinary_repeat_waits_for_running_worker_and_checks_consumer(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1) + self.execute(cluster, folder) + self.assertIn(("rollout", "worker"), cluster.events) + self.assertIn(("broker-consumer-check",), cluster.events) + + def test_ordinary_repeat_rejects_failed_worker_rollout(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1, worker_error=True) + with self.assertRaisesRegex(RuntimeError, "worker rollout failed"): + self.execute(cluster, folder) + self.assertNotIn(("broker-consumer-check",), cluster.events) + self.assertEqual(cluster.replicas["worker"], 0) + + def test_ordinary_repeat_stops_worker_when_consumer_is_missing(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1, consumer_count=0) + with self.assertRaisesRegex(RuntimeError, "exactly one"): + self.execute(cluster, folder) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertEqual(cluster.events.count(("broker-consumer-check",)), 30) + + def test_enabling_worker_waits_for_delayed_consumer(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(consumer_count=[0, 0, 1]) + self.execute(cluster, folder, enable=True, verified=True) + self.assertEqual(cluster.replicas["worker"], 1) + self.assertEqual(cluster.events.count(("broker-consumer-check",)), 3) + self.assertLess(cluster.events.index(("rollout", "worker")), cluster.events.index(("broker-consumer-check",))) + + def test_ordinary_repeat_waits_for_delayed_consumer(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1, consumer_count=[0, 1]) + self.execute(cluster, folder) + self.assertEqual(cluster.replicas["worker"], 1) + self.assertEqual(cluster.events.count(("broker-consumer-check",)), 2) + + def test_extra_consumer_stops_worker_immediately(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(consumer_count=[2, 1]) + with self.assertRaisesRegex(RuntimeError, "unexpected broker count"): + self.execute(cluster, folder, enable=True, verified=True) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertEqual(cluster.events.count(("broker-consumer-check",)), 1) + + def test_consumer_gate_zero_timeout_performs_no_broker_query(self): + config, _ = fixture() + cluster = FakeCluster() + with self.assertRaisesRegex(RuntimeError, "before timeout"): + ns.wait_for_worker_consumer(cluster, config, timeout=0) + self.assertEqual(cluster.events, []) + + def test_consumer_query_has_bounded_timeout_and_redacts_failure(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1) + with patch.object(cluster, "run", side_effect=RuntimeError("private broker detail")) as run: + with self.assertRaisesRegex(RuntimeError, "could not verify") as error: + self.execute(cluster, folder) + self.assertNotIn("private broker detail", str(error.exception)) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertEqual(run.call_args.kwargs, {"timeout": 20, "sensitive": True}) + + def test_consumer_query_cannot_succeed_after_deadline(self): + config, _ = fixture() + with patch.object(ns.time, "monotonic", side_effect=[0, 0, 61]): + with self.assertRaisesRegex(RuntimeError, "before timeout"): + ns.wait_for_worker_consumer(FakeCluster(), config) + + def test_backup_failure_leaves_both_writers_stopped(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1) + with self.assertRaises(RuntimeError): + self.execute(cluster, folder, migrate=True, backup_error=True) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertEqual(cluster.replicas["web"], 0) + self.assertNotIn(("apply",), cluster.events) + + def test_migration_failure_leaves_both_writers_stopped(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1) + with self.assertRaises(RuntimeError): + self.execute(cluster, folder, migrate=True, job_error=True) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertEqual(cluster.replicas["web"], 0) + self.assertLess(cluster.events.index(("backup",)), cluster.events.index(("job", "notification-migrator"))) + + def test_noop_migration_resumes_web_after_grants_and_keeps_worker_stopped(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1) + self.execute(cluster, folder, migrate=True) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertEqual(cluster.replicas["web"], 1) + self.assertLess(cluster.events.index(("job", "notification-grants")), cluster.events.index(("scale", "web", 1))) + self.assertLess(cluster.events.index(("job", "notification-sandbox")), cluster.events.index(("scale", "web", 1))) + + def test_unknown_history_stops_before_mutation(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(history=["unknown"]) + with self.assertRaises(RuntimeError): + self.execute(cluster, folder, migrate=True) + self.assertEqual(cluster.events, [("history",)]) + + def test_interrupted_first_install_recovers_storage_before_history_then_backs_up(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(history=[], worker=0, postgres_ready=False) + self.execute(cluster, folder, migrate=True) + self.assertLess(cluster.events.index(("rollout", "postgres")), cluster.events.index(("history",))) + self.assertLess(cluster.events.index(("history",)), cluster.events.index(("backup",))) + self.assertLess(cluster.events.index(("backup",)), cluster.events.index(("job", "notification-migrator"))) + first_deployments = {v["metadata"]["name"] for v in cluster.applications[0] if v["kind"] == "Deployment"} + self.assertEqual(first_deployments, {"postgres", "rabbitmq", "mailpit"}) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertEqual(cluster.replicas["web"], 1) + + def test_missing_deployment_does_not_skip_backup_of_retained_database(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(postgres_exists=False, postgres_ready=False) + self.execute(cluster, folder, migrate=True) + self.assertIn(("backup",), cluster.events) + self.assertLess(cluster.events.index(("backup",)), cluster.events.index(("job", "notification-migrator"))) + + def test_recovery_backup_failure_keeps_writers_stopped_and_never_migrates(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1, postgres_ready=False) + with self.assertRaisesRegex(RuntimeError, "backup failed"): + self.execute(cluster, folder, migrate=True, backup_error=True) + self.assertEqual(cluster.replicas["web"], 0) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertEqual(len(cluster.applications), 1) # Foundation only. + self.assertFalse(any(event[0] == "job" for event in cluster.events)) + + def test_recovery_storage_failure_does_not_query_history_or_start_writers(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(worker=1, postgres_ready=False, storage_error=True) + with self.assertRaisesRegex(RuntimeError, "postgres rollout failed"): + self.execute(cluster, folder, migrate=True) + self.assertNotIn(("history",), cluster.events) + self.assertNotIn(("backup",), cluster.events) + self.assertEqual(cluster.replicas["web"], 0) + self.assertEqual(cluster.replicas["worker"], 0) + + def test_recovery_unknown_history_keeps_writers_stopped_and_never_migrates(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(history=["unknown"], worker=1, postgres_ready=False) + with self.assertRaisesRegex(RuntimeError, "unknown or non-prefix"): + self.execute(cluster, folder, migrate=True) + self.assertEqual(cluster.replicas["web"], 0) + self.assertEqual(cluster.replicas["worker"], 0) + self.assertFalse(any(event[0] == "job" for event in cluster.events)) + + def test_ordinary_deploy_of_stopped_database_requires_recovery_without_mutation(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster(postgres_ready=False) + with self.assertRaisesRegex(RuntimeError, "recovery requires --migrate"): + self.execute(cluster, folder) + self.assertEqual(cluster.events, []) + + def test_enabling_worker_without_inspection_fails_before_mutation(self): + with tempfile.TemporaryDirectory() as folder: + cluster = FakeCluster() + with self.assertRaises(RuntimeError): + self.execute(cluster, folder, enable=True) + self.assertEqual(cluster.events, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/docker/kubernetes/shared-notification/tests/test_seq_key.py b/docker/kubernetes/shared-notification/tests/test_seq_key.py new file mode 100644 index 0000000..2166c5f --- /dev/null +++ b/docker/kubernetes/shared-notification/tests/test_seq_key.py @@ -0,0 +1,317 @@ +import copy +import importlib.util +import json +from pathlib import Path +import tempfile +import unittest + + +PACKAGE = Path(__file__).resolve().parents[1] +spec = importlib.util.spec_from_file_location("prepare_seq_key", PACKAGE / "prepare-seq-key.py") +seq = importlib.util.module_from_spec(spec) +spec.loader.exec_module(seq) + + +def key(token="a" * 64): + return {"Id": "apikey-notification", "Title": seq.TITLE, "Token": token, + "AssignedPermissions": ["Ingest"], "InputSettings": {"AppliedProperties": copy.deepcopy(seq.PROPERTIES)}} + + +class FakeSeq: + def __init__(self, keys=None, token_file=None, fail_after_create=False, fail_before_put=False, fail_after_put=False): + self.keys = copy.deepcopy(keys or []) + self.calls = [] + self.token_file = token_file + self.fail_after_create = fail_after_create + self.events = [] + self.fail_before_put = fail_before_put + self.fail_after_put = fail_after_put + + def request(self, method, path, body=None, key=None): + self.calls.append((method, path)) + if method == "GET" and path in ("api/apikeys", seq.KEY_LIST): + # Actual API may omit Token: probe must verify it independently. + visible = self.keys if path == seq.KEY_LIST else [v for v in self.keys if v.get("OwnerId") is not None] + return [{k: copy.deepcopy(v) for k, v in value.items() if k != "Token"} for value in visible] + if path == "api/apikeys/template": + return {"InputSettings": {"AppliedProperties": []}} + if path == "api/apikeys/": + self.assert_saved_before_post(body["Token"]) + value = copy.deepcopy(body) + value["Id"] = "apikey-notification" + self.keys.append(value) + if self.fail_after_create: + raise RuntimeError("trusted Seq endpoint unavailable") + return copy.deepcopy(value) + if method == "PUT" and path.startswith("api/apikeys/"): + if self.fail_before_put: + raise RuntimeError("trusted Seq endpoint unavailable") + index = next(i for i, value in enumerate(self.keys) if path == "api/apikeys/" + value["Id"]) + original_token = self.keys[index].get("Token") + self.keys[index] = copy.deepcopy(body) + self.keys[index]["Token"] = original_token + if self.fail_after_put: + raise RuntimeError("trusted Seq endpoint unavailable") + return copy.deepcopy(self.keys[index]) + if path == "api/events/raw": + origin = next((value for value in self.keys if value.get("Token") == key), None) + if origin is None: + raise RuntimeError("Seq request denied (HTTP 401)") + canary = json.loads(body) + if set(canary) != {"@t", "@mt", "Application", "ProvisioningCheckId"}: + raise AssertionError("unexpected canary scope") + properties = {k: v for k, v in canary.items() if not k.startswith("@")} + properties.update({p["Name"]: p["Value"] for p in origin["InputSettings"]["AppliedProperties"]}) + self.events.append({"ApiKeyId": origin["Id"], "Properties": [{"Name": k, "Value": v} for k, v in properties.items()]}) + return None + if method == "GET" and path.startswith("api/events?count=2&filter="): + query = seq.urllib.parse.unquote(path.split("filter=", 1)[1]) + check_id = query.split("'", 2)[1] + return [event for event in self.events if any(p == {"Name": "ProvisioningCheckId", "Value": check_id} for p in event["Properties"])] + raise AssertionError("unexpected API operation") + + def assert_saved_before_post(self, token): + if self.token_file is None or self.token_file.read_text().strip() != token: + raise AssertionError("token was not persisted before create") + + +class SeqKeyTests(unittest.TestCase): + def test_create_preserves_token_before_post_and_does_not_touch_staging_key(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + staging = {"Id": "apikey-staging", "Title": "invemory-staging-ingest", "AssignedPermissions": ["Ingest"]} + client = FakeSeq([staging], path) + self.assertEqual(seq.provision(client, path), "apikey-notification") + self.assertEqual(client.keys[0], staging) + self.assertEqual(client.keys[1]["AssignedPermissions"], ["Ingest"]) + self.assertEqual(client.keys[1]["InputSettings"]["AppliedProperties"], seq.PROPERTIES + [{"Name": seq.MARKER, "Value": "apikey-notification"}]) + self.assertEqual(path.read_text().strip(), client.keys[1]["Token"]) + + def test_repeat_verifies_existing_key_without_rotation(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + client = FakeSeq(token_file=path) + seq.provision(client, path) + original = path.read_bytes() + client.calls.clear() + seq.provision(client, path) + self.assertEqual(path.read_bytes(), original) + self.assertEqual(len(client.keys), 1) + self.assertNotIn(("POST", "api/apikeys/"), client.calls) + self.assertFalse(any(method == "PUT" for method, _ in client.calls)) + self.assertTrue(all(seq.allowed_path(method, path) for method, path in client.calls)) + + def test_ambiguous_keys_fail_before_token_generation(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + with self.assertRaisesRegex(RuntimeError, "ambiguous"): + seq.provision(FakeSeq([key(), key()]), path) + self.assertFalse(path.exists()) + + def test_existing_key_with_extra_permissions_is_not_modified(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + existing = key() + existing["AssignedPermissions"].append("Read") + client = FakeSeq([existing]) + with self.assertRaisesRegex(RuntimeError, "only Ingest"): + seq.provision(client, path) + self.assertFalse(path.exists()) + self.assertEqual(client.keys, [existing]) + + def test_existing_key_with_wrong_stamp_is_not_modified(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + existing = key() + existing["InputSettings"]["AppliedProperties"][0]["Value"] = "Staging" + client = FakeSeq([existing]) + with self.assertRaisesRegex(RuntimeError, "properties mismatch"): + seq.provision(client, path) + self.assertFalse(path.exists()) + self.assertEqual(client.keys, [existing]) + + def test_missing_local_token_does_not_rotate_remote_key(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + client = FakeSeq([key()]) + with self.assertRaisesRegex(RuntimeError, "local token is missing"): + seq.provision(client, path) + self.assertFalse(path.exists()) + self.assertEqual(client.calls, [("GET", seq.KEY_LIST)]) + + def test_unknown_post_outcome_is_recovered_with_original_token(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + client = FakeSeq(token_file=path, fail_after_create=True) + with self.assertRaises(RuntimeError): + seq.provision(client, path) + original = path.read_bytes() + client.fail_after_create = False + seq.provision(client, path) + self.assertEqual(path.read_bytes(), original) + self.assertEqual(len(client.keys), 1) + + def test_wrong_preserved_token_fails_probe_without_rotation(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + seq.private_directory(path.parent) + path.write_text("b" * 64) + path.chmod(0o600) + client = FakeSeq([key()]) + with self.assertRaisesRegex(RuntimeError, "HTTP 401"): + seq.provision(client, path) + self.assertEqual(path.read_text(), "b" * 64) + self.assertNotIn(("POST", "api/apikeys/"), client.calls) + + def test_valid_foreign_staging_token_is_rejected_by_canary_identity(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + seq.private_directory(path.parent) + path.write_text("b" * 64) + path.chmod(0o600) + staging = key("b" * 64) + staging.update(Id="apikey-staging", Title="invemory-staging-ingest") + staging["InputSettings"]["AppliedProperties"] = [{"Name": "Environment", "Value": "Staging"}] + client = FakeSeq([key(), staging]) + with self.assertRaisesRegex(RuntimeError, "different or unverified key"): + seq.provision(client, path) + self.assertIn(("POST", "api/events/raw"), client.calls) # Valid ingestion alone is insufficient. + self.assertNotIn(("POST", "api/apikeys/"), client.calls) + self.assertEqual(path.read_text(), "b" * 64) + self.assertEqual(client.keys[1], staging) + self.assertEqual(client.keys[0]["Token"], key()["Token"]) + + def test_foreign_key_with_matching_properties_is_still_rejected(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + seq.private_directory(path.parent) + path.write_text("b" * 64) + path.chmod(0o600) + foreign = key("b" * 64) + foreign.update(Id="apikey-foreign", Title="Other Service") + with self.assertRaisesRegex(RuntimeError, "different or unverified key"): + seq.provision(FakeSeq([key(), foreign]), path) + + def test_server_stamped_marker_proves_identity_without_event_key_id_or_token(self): + client = FakeSeq() + check_id = "c" * 32 + client.events = [{"Properties": [{"Name": "ProvisioningCheckId", "Value": check_id}, + {"Name": seq.MARKER, "Value": "apikey-notification"}] + copy.deepcopy(seq.PROPERTIES)}] + redacted = key() + redacted["Token"] = None + seq.verify_canary(client, check_id, redacted) + + def test_missing_server_stamped_marker_fails_closed(self): + client = FakeSeq() + check_id = "c" * 32 + client.events = [{"Properties": [{"Name": "ProvisioningCheckId", "Value": check_id}] + copy.deepcopy(seq.PROPERTIES)}] + redacted = key() + redacted.pop("Token") + with self.assertRaisesRegex(RuntimeError, "unverified key"): + seq.verify_canary(client, check_id, redacted) + + def test_correct_key_id_still_requires_actual_shared_stamps(self): + client = FakeSeq() + check_id = "c" * 32 + client.events = [{"ApiKeyId": "apikey-notification", "Properties": [ + {"Name": "ProvisioningCheckId", "Value": check_id}, {"Name": seq.MARKER, "Value": "apikey-notification"}, + {"Name": "Environment", "Value": "Staging"}]}] + with self.assertRaisesRegex(RuntimeError, "stamped identity mismatch"): + seq.verify_canary(client, check_id, key()) + + def test_marker_collision_fails_before_put_or_ingestion(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + seq.private_directory(path.parent) + path.write_text("a" * 64) + path.chmod(0o600) + foreign = key("b" * 64) + foreign.update(Id="apikey-foreign", Title="Other Service") + foreign["InputSettings"]["AppliedProperties"].append({"Name": seq.MARKER, "Value": "apikey-notification"}) + client = FakeSeq([key(), foreign]) + with self.assertRaisesRegex(RuntimeError, "marker collision"): + seq.provision(client, path) + self.assertFalse(any(method == "PUT" for method, _ in client.calls)) + self.assertNotIn(("POST", "api/events/raw"), client.calls) + self.assertEqual(client.keys, [key(), foreign]) + + def test_shared_foreign_marker_collision_is_visible_in_full_listing(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + seq.private_directory(path.parent) + path.write_text("a" * 64) + path.chmod(0o600) + dedicated = key() + dedicated["OwnerId"] = "user-operator" + shared = key("b" * 64) + shared.update(Id="apikey-shared-foreign", Title="Shared foreign service", OwnerId=None) + shared["InputSettings"]["AppliedProperties"].append({"Name": seq.MARKER, "Value": dedicated["Id"]}) + client = FakeSeq([dedicated, shared]) + # Demonstrate the narrower default endpoint would conceal the collision. + self.assertEqual(len(client.request("GET", "api/apikeys")), 1) + client.calls.clear() + with self.assertRaisesRegex(RuntimeError, "marker collision"): + seq.provision(client, path) + self.assertTrue(all(call == ("GET", seq.KEY_LIST) for call in client.calls)) + self.assertEqual(client.keys, [dedicated, shared]) + self.assertEqual(path.read_text(), "a" * 64) + + def test_allowlist_requires_full_shared_key_listing(self): + self.assertTrue(seq.allowed_path("GET", "api/apikeys?shared=true")) + self.assertFalse(seq.allowed_path("GET", "api/apikeys")) + self.assertFalse(seq.allowed_path("GET", "api/apikeys?shared=false")) + + def test_guarded_put_preserves_token_permissions_and_all_other_settings(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + seq.private_directory(path.parent) + path.write_text("a" * 64) + path.chmod(0o600) + original = key() + original.update(OwnerId=None, IsDefault=False, TokenPrefix="aaaa") + original["InputSettings"].update(MinimumLevel="Information", Filter=None, CustomSettings={"retained": True}) + expected = copy.deepcopy(original) + expected["InputSettings"]["AppliedProperties"].append({"Name": seq.MARKER, "Value": "apikey-notification"}) + client = FakeSeq([original]) + seq.provision(client, path) + self.assertEqual(client.keys, [expected]) + self.assertEqual(path.read_text(), "a" * 64) + self.assertEqual([call for call in client.calls if call[0] == "PUT"], [("PUT", "api/apikeys/apikey-notification")]) + + def test_put_failure_preserves_original_key_and_local_token(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + seq.private_directory(path.parent) + path.write_text("a" * 64) + path.chmod(0o600) + client = FakeSeq([key()], fail_before_put=True) + with self.assertRaisesRegex(RuntimeError, "unavailable"): + seq.provision(client, path) + self.assertEqual(client.keys, [key()]) + self.assertEqual(path.read_text(), "a" * 64) + self.assertNotIn(("POST", "api/events/raw"), client.calls) + + def test_unknown_put_outcome_is_recovered_without_another_put_or_rotation(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder) / "private" / "seq-key.txt" + seq.private_directory(path.parent) + path.write_text("a" * 64) + path.chmod(0o600) + client = FakeSeq([key()], fail_after_put=True) + with self.assertRaisesRegex(RuntimeError, "unavailable"): + seq.provision(client, path) + client.calls.clear() + client.fail_after_put = False + seq.provision(client, path) + self.assertFalse(any(method == "PUT" for method, _ in client.calls)) + self.assertNotIn(("POST", "api/apikeys/"), client.calls) + self.assertEqual(path.read_text(), "a" * 64) + self.assertEqual(len(client.keys), 1) + + def test_redirect_is_rejected_before_credentials_can_be_forwarded(self): + with self.assertRaisesRegex(RuntimeError, "redirect denied"): + seq.NoRedirects().redirect_request(None, None, 302, "Found", {}, "https://untrusted.example/") + + +if __name__ == "__main__": + unittest.main()