hrynco/hrynco-notification-service
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request 'chore: preserve current configuration and documentation' (#20) from feature/it-1493-notification-k8s into main

Browse Source
This commit was merged in pull request #20.
This commit is contained in:
Anatolii Hrynchuk
2026-10-03 00:59:44 +03:00
parent a44b98292f 27f3603d8f
commit a684caf7f7
11 changed files with 2320 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.dockerignore
+10 -9
View File
@@ -1,9 +1,10 @@
**/.git **/.git
**/.vs **/.vs
**/.idea **/.idea
**/bin **/bin
**/obj **/obj
**/*.user **/*.user
**/*.suo **/*.suo
**/TestResults **/TestResults
NuGet.Config NuGet.Config
docker/kubernetes/
README.md
+121 -120
View File
@@ -1,120 +1,121 @@
# hrynco-notification-service # 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. 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 ## Documentation
- [ItemTracker outbox email consumer](docs/itemtracker-outbox-email-consumer.md) - [ItemTracker outbox email consumer](docs/itemtracker-outbox-email-consumer.md)
- [Shared Notification Service Kubernetes package](docker/kubernetes/shared-notification/README.md)
## Development environment
## Development environment
The development Docker Compose stack runs PostgreSQL, RabbitMQ, database migrations,
the Notification Service Web and Worker applications, Seq, and Mailpit. 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: Prerequisite: install and start Docker Desktop. Then run the installation script from
the repository root:
```powershell
.\docker\environments\install-development.cmd ```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 script validates Docker and the Compose configuration, builds the application
the main development URLs. It can also be launched directly from File Explorer or images, starts the complete stack in the background, and prints container status and
from another working directory. 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:
To validate the setup without building images or creating containers, run:
```powershell
.\docker\environments\install-development.cmd --validate-only ```powershell
``` .\docker\environments\install-development.cmd --validate-only
```
By default, the script uses the tracked `docker/environments/.env.Development` file:
By default, the script uses the tracked `docker/environments/.env.Development` file:
```dotenv
DB_NAME=notification_service ```dotenv
DB_USER=postgres DB_NAME=notification_service
DB_PASS=postgres DB_USER=postgres
VOLUME_PREFIX=ns-dev DB_PASS=postgres
RABBITMQ_USER=guest VOLUME_PREFIX=ns-dev
RABBITMQ_PASSWORD=guest RABBITMQ_USER=guest
RABBITMQ_AMQP_PORT=5672 RABBITMQ_PASSWORD=guest
RABBITMQ_MANAGEMENT_PORT=15672 RABBITMQ_AMQP_PORT=5672
WEB_PORT=5200 RABBITMQ_MANAGEMENT_PORT=15672
DEVELOPMENT_EMAIL_SERVICE_NAME=TestService 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: 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 ` ```powershell
-f docker/environments/docker-compose.yml ` docker compose --env-file docker/environments/.env.local `
-f docker/environments/docker-compose.Development.yml ` -f docker/environments/docker-compose.yml `
up --build -d -f docker/environments/docker-compose.Development.yml `
``` up --build -d
```
After startup, the main development endpoints are:
After startup, the main development endpoints are:
- Notification Service admin: `http://localhost:5200/admin/channels`
- Notification templates: `http://localhost:5200/admin/templates` - Notification Service admin: `http://localhost:5200/admin/channels`
- RabbitMQ management: `http://localhost:15672` (`guest` / `guest`) - Notification templates: `http://localhost:5200/admin/templates`
- Mailpit inbox: `http://localhost:8025` - RabbitMQ management: `http://localhost:15672` (`guest` / `guest`)
- Seq logs: `http://localhost:5342` - Mailpit inbox: `http://localhost:8025`
- PostgreSQL: `localhost:5433` - 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 During development migrations, the stack idempotently creates an active Mailpit SMTP
`DEVELOPMENT_EMAIL_SERVICE_NAME`. Existing channels and templates are never channel and a neutral English `TestEmail` template for
overwritten. Change that environment value when the publisher uses another service `DEVELOPMENT_EMAIL_SERVICE_NAME`. Existing channels and templates are never
name. overwritten. Change that environment value when the publisher uses another service
name.
Check container status and follow Worker logs:
Check container status and follow Worker logs:
```powershell
docker compose --env-file docker/environments/.env.local ` ```powershell
-f docker/environments/docker-compose.yml ` docker compose --env-file docker/environments/.env.local `
-f docker/environments/docker-compose.Development.yml ` -f docker/environments/docker-compose.yml `
ps -f docker/environments/docker-compose.Development.yml `
ps
docker compose --env-file docker/environments/.env.local `
-f docker/environments/docker-compose.yml ` docker compose --env-file docker/environments/.env.local `
-f docker/environments/docker-compose.Development.yml ` -f docker/environments/docker-compose.yml `
logs -f worker -f docker/environments/docker-compose.Development.yml `
``` logs -f worker
```
Stop the environment without deleting its database and RabbitMQ volumes:
Stop the environment without deleting its database and RabbitMQ volumes:
```powershell
docker compose --env-file docker/environments/.env.local ` ```powershell
-f docker/environments/docker-compose.yml ` docker compose --env-file docker/environments/.env.local `
-f docker/environments/docker-compose.Development.yml ` -f docker/environments/docker-compose.yml `
down -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. 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
## Notification worker flow
```mermaid
flowchart TD ```mermaid
A[Worker host starts] --> B[Load config and register services] flowchart TD
B --> C[Start SendEmailConsumer] A[Worker host starts] --> B[Load config and register services]
C --> D[Receive message from notification.send-email] B --> C[Start SendEmailConsumer]
D --> E[Validate Notification.SendEmail.v1 metadata and payload] C --> D[Receive message from notification.send-email]
E --> F[Resolve SendEmailService] D --> E[Validate Notification.SendEmail.v1 metadata and payload]
F --> G[Pick channel and exact-language template] E --> F[Resolve SendEmailService]
G --> H[Validate variables and render email content] F --> G[Pick channel and exact-language template]
H --> I[Send via SMTP] G --> H[Validate variables and render email content]
I --> J[Update usage counters] H --> I[Send via SMTP]
J --> K[Optionally publish result to reply queue] I --> J[Update usage counters]
K --> L[Acknowledge RabbitMQ delivery] J --> K[Optionally publish result to reply queue]
I -. failure .-> M[Log and retry] K --> L[Acknowledge RabbitMQ delivery]
M -->|retries exhausted| N[Publish terminal failure result] I -. failure .-> M[Log and retry]
N --> O[Nack original delivery without requeue] 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. Template variables are treated as plain text. The worker HTML-encodes every variable while
Templates must express markup in `body.html` instead of supplying HTML through variables. 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.
docker/kubernetes/shared-notification/.gitignore
+3
View File
@@ -0,0 +1,3 @@
*.local.json
private/
rendered/
docker/kubernetes/shared-notification/README.md
+331
View File
@@ -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=<dedicated key ID>` 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.
docker/kubernetes/shared-notification/artifacts.example.json
+14
View File
@@ -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"}
}
}
docker/kubernetes/shared-notification/config.example.json
+19
View File
@@ -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"
}
docker/kubernetes/shared-notification/notification.py
+695
View File
@@ -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',
'<p>Hello {{RecipientName}},</p><p>{{Message}}</p>', '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)
docker/kubernetes/shared-notification/prepare-seq-key.py
+276
View File
@@ -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)
docker/kubernetes/shared-notification/prepare-storage.py
Executable
+88
View File
@@ -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)
docker/kubernetes/shared-notification/tests/test_notification.py
+446
View File
@@ -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()
docker/kubernetes/shared-notification/tests/test_seq_key.py
+317
View File
@@ -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()