From 188314922dc96b68ceb56a1fc52d47fcd41f2dfd Mon Sep 17 00:00:00 2001 From: edwardvaneechoud Date: Wed, 13 May 2026 20:54:23 +0200 Subject: [PATCH 1/3] Improving secret management and adding documentation about effects of --- docs/for-developers/secret-storage.md | 170 ++++++++++++ docs/users/visual-editor/catalog/secrets.md | 177 ++++++++++-- .../versions/012_secret_access_events.py | 64 +++++ flowfile_core/flowfile_core/auth/secrets.py | 27 ++ .../flowfile_core/database/models.py | 34 +++ flowfile_core/flowfile_core/routes/public.py | 10 +- flowfile_core/flowfile_core/routes/secrets.py | 259 ++++++++++++++---- .../flowfile_core/secret_manager/audit.py | 140 ++++++++++ .../secret_manager/secret_manager.py | 143 +++------- flowfile_core/tests/test_secret_audit.py | 159 +++++++++++ flowfile_worker/flowfile_worker/secrets.py | 155 ++++------- mkdocs.yml | 1 + shared/crypto/__init__.py | 26 ++ shared/crypto/envelope.py | 96 +++++++ 14 files changed, 1183 insertions(+), 278 deletions(-) create mode 100644 docs/for-developers/secret-storage.md create mode 100644 flowfile_core/flowfile_core/alembic/versions/012_secret_access_events.py create mode 100644 flowfile_core/flowfile_core/secret_manager/audit.py create mode 100644 flowfile_core/tests/test_secret_audit.py create mode 100644 shared/crypto/__init__.py create mode 100644 shared/crypto/envelope.py diff --git a/docs/for-developers/secret-storage.md b/docs/for-developers/secret-storage.md new file mode 100644 index 000000000..a1b439a6e --- /dev/null +++ b/docs/for-developers/secret-storage.md @@ -0,0 +1,170 @@ +# Secret Storage + +Reference for developers touching Flowfile's secret-storage subsystem. For the +operator's manual on backup, recovery, and threat scenarios, see +[`users/visual-editor/catalog/secrets.md`](../users/visual-editor/catalog/secrets.md). + +## Where things live + +| Concern | File | +|---|---| +| Envelope format + crypto primitives | `shared/crypto/envelope.py` | +| Master-key fetch (core, with auto-generation) | `flowfile_core/flowfile_core/auth/secrets.py` | +| Master-key fetch (worker) | `flowfile_worker/flowfile_worker/secrets.py` | +| High-level wrappers (`encrypt_secret`, `decrypt_secret`, `store_secret`) | `flowfile_core/flowfile_core/secret_manager/secret_manager.py` | +| Worker wrappers (same surface, worker-side master-key lookup) | `flowfile_worker/flowfile_worker/secrets.py` | +| Audit helper | `flowfile_core/flowfile_core/secret_manager/audit.py` | +| Audit table model | `SecretAccessEvent` in `flowfile_core/flowfile_core/database/models.py` | +| Audit table migration | `flowfile_core/flowfile_core/alembic/versions/012_secret_access_events.py` | +| CRUD routes (audit-wired) | `flowfile_core/flowfile_core/routes/secrets.py` | + +Both `flowfile_core` and `flowfile_worker` delegate the actual crypto to +`shared/crypto/envelope.py`. Each service owns its own master-key resolution +(different backends) but they agree byte-for-byte on the envelope. + +## Envelope format + +``` +$ffsec$1$$ +``` + +Stored as plain text in `secrets.encrypted_value`. The `` is embedded +so the worker can decrypt without separately being told whose secret it is. +The leading `1` is a version digit — see [Adding envelope v2](#adding-envelope-v2). + +Legacy raw Fernet tokens (no `$ffsec$` prefix) are still readable via +`decrypt_secret_envelope(..., user_id=...)` for backwards compatibility. + +## Shared crypto API + +`from shared.crypto.envelope import ...` + +| Symbol | Purpose | +|---|---| +| `KEY_DERIVATION_VERSION` | HKDF salt — `b"flowfile-secrets-v1"`. Changing this invalidates every existing secret. | +| `SECRET_FORMAT_PREFIX` | `"$ffsec$1$"` | +| `derive_user_key(master_key, user_id)` | HKDF-SHA256 → 32-byte URL-safe base64 Fernet key. Deterministic. | +| `parse_v1_envelope(blob) → (user_id, token)` | Splits a v1 envelope. Raises `ValueError` on malformed input. | +| `encrypt_secret_envelope(master_key, plaintext, user_id)` | Emit a v1 envelope. | +| `decrypt_secret_envelope(master_key, blob, user_id=None)` | Decrypt a v1 envelope (user_id read from blob) or a legacy raw token. | + +Don't import `shared.crypto.envelope` directly from a route handler. Use the +`secret_manager` wrappers — they fetch the master key locally and delegate. + +```python +from flowfile_core.secret_manager.secret_manager import encrypt_secret, decrypt_secret + +token = encrypt_secret("hunter2", user_id=current_user.id) +plaintext = decrypt_secret(token).get_secret_value() +``` + +## Master-key resolution + +**Core** (`flowfile_core/auth/secrets.py::get_master_key`) + +1. `FLOWFILE_MODE=docker` → `get_docker_secret_key()` reads + `FLOWFILE_MASTER_KEY` env var or `/run/secrets/flowfile_master_key`. + Raises `RuntimeError` if neither is configured. +2. Otherwise → `SecureStorage.get_password("flowfile", "master_key")`. + Auto-generates on first call and persists to + `/flowfile.json.enc` encrypted under `/.secret_key`. + The auto-generation branch emits a multi-line `logger.warning(...)` with + both file paths and an 8-char SHA-256 fingerprint. Fires once. + +**Worker** (`flowfile_worker/secrets.py::get_master_key`) + +1. `TEST_MODE=1` → use `FLOWFILE_MASTER_KEY` if set, else generate a fresh + per-process random key cached in module-level `_TEST_MASTER_KEY`. +2. `FLOWFILE_MODE=docker` → same Docker resolution as core. +3. Otherwise → read from worker's local `SecureStorage`. Worker does **not** + auto-generate; it raises `ValueError` if the key is missing. (Core is + expected to be the generation point in non-Docker deployments; worker + reads what core wrote.) + +In production, core and worker share the key by both reading the same +`FLOWFILE_MASTER_KEY` env var (Docker) or the same on-disk +`flowfile.json.enc` (local install). + +## Audit log + +Table: `secret_access_events`. + +| Column | Type | Notes | +|---|---|---| +| `id` | int PK | | +| `user_id` | FK → `users.id`, indexed | | +| `secret_id` | FK → `secrets.id` ON DELETE SET NULL | preserves audit row after deletion | +| `secret_name` | string, indexed, nullable | denormalized — survives the FK going NULL | +| `action` | string, indexed | `create` / `list` / `read` / `delete` | +| `result_status` | string | `success` / `error` | +| `error` | text, nullable | short snake_case codes: `not_found`, `duplicate_name`, … | +| `source` | string, default `api` | reserved for future non-API emitters | +| `ip_address` | string, nullable | best-effort; honors `X-Forwarded-For` | +| `created_at` | datetime, indexed | server-side `now()` | + +### Recording an event from a new route + +```python +from flowfile_core.secret_manager import audit + +audit.record_event( + audit.SecretEvent( + user_id=current_user.id, + action="read", + result_status="success", + secret_name=name, + secret_id=row.id, + ip_address=request.client.host if request.client else None, + ), + db=db, +) +db.commit() +``` + +`record_event` adds the row to the caller's session — the caller is responsible +for committing. Exceptions during the audit insert are logged and swallowed +rather than failing the user-facing operation that already succeeded. + +### What is *not* recorded + +Per-secret decrypt activity during flow execution. The audit log is for the +API CRUD boundary. Logging every decrypt would dominate the table with +routine flow-run activity. If per-decrypt accounting becomes load-bearing, +add a separate event source rather than expanding this one. + +## Test mode + +`TEST_MODE=1` (set by `flowfile_worker/tests/conftest.py`) makes +`get_master_key()` synthesize a per-process random key when +`FLOWFILE_MASTER_KEY` isn't set. Round-trip tests work without external setup. + +Integration tests that need core and worker to agree on the key must export +`FLOWFILE_MASTER_KEY` explicitly — see +`flowfile_core/tests/test_auth_e2e.py` for the pattern. + +No hardcoded key material lives in source. + +## Adding envelope v2 + +The dispatch in `decrypt_secret_envelope` is set up to branch on the version +digit: + +```python +if encrypted_value.startswith(SECRET_FORMAT_PREFIX): # "$ffsec$1$" + ... +elif encrypted_value.startswith(SECRET_FORMAT_PREFIX_V2): # "$ffsec$2$" (add this) + ... +``` + +To introduce a v2: + +1. Add `SECRET_FORMAT_PREFIX_V2 = "$ffsec$2$"` in `envelope.py`. +2. Add `parse_v2_envelope` and a v2 decrypt branch. +3. Flip `encrypt_secret_envelope` to emit v2 (or split write paths). +4. Decide a re-write policy for existing v1 rows: lazy on read, an explicit + admin-triggered migrate, or leave both formats forever. The contract is + that v1 must remain decryptable — never break old envelopes. + +The natural shape for KEK-rotation support is +`$ffsec$2$$$` plus a `master_key_versions` table +tracking which KEK ID is active; old envelopes get re-wrapped lazily on read. diff --git a/docs/users/visual-editor/catalog/secrets.md b/docs/users/visual-editor/catalog/secrets.md index ce80464af..736c70c38 100644 --- a/docs/users/visual-editor/catalog/secrets.md +++ b/docs/users/visual-editor/catalog/secrets.md @@ -1,47 +1,91 @@ # Secrets -Store sensitive credentials like database passwords and API keys securely. +Store sensitive credentials — database passwords, cloud connection keys, API tokens — outside your flow definitions. Flowfile encrypts them at rest and decrypts them on demand at runtime. -## How It Works +This page is the operator's manual. It explains exactly what's encrypted, where the master key lives, how to back it up, and what to do when something goes wrong. -Secrets are encrypted using a master key before storage. When a flow needs a credential, Flowfile decrypts it on-demand. The actual values never appear in flow definitions or logs. +## What gets encrypted -## Master Key +Anything you enter into Flowfile that wouldn't be safe to print: -The master key encrypts all secrets. Without it, secrets cannot be decrypted. +- Database connection passwords (PostgreSQL, MySQL, SQL Server, …) +- Cloud connection credentials — AWS secret keys, Azure account keys / SAS tokens / client secrets, GCS service account JSON +- OAuth refresh tokens (Google Analytics, etc.) +- Kafka SASL passwords and SSL keys +- AI provider API keys (the "BYOK" credentials in the AI panel) -### Configuration by Mode +**What's *not* encrypted:** the *names* you give your secrets and the *metadata* of your connections (host, port, username, etc.). Treat connection names as non-sensitive. -| Mode | Configuration | -|------|---------------| -| **Desktop (Electron)** | Auto-generated on first open, stored at `~/.config/flowfile/` | -| **Python API** | Auto-generated on first use, stored at `~/.config/flowfile/` | -| **Docker** | Generate via setup wizard, set as `FLOWFILE_MASTER_KEY` env variable | +## How the encryption works -### Desktop & Python API +Three short ideas, in order. -The master key is automatically generated on first use and stored securely. No manual configuration needed. +**Master key.** One Fernet key per Flowfile instance. Every encrypted secret in your database is unlockable by this key — and only this key. If it's lost, the encrypted secrets are unrecoverable; there is no backdoor. -!!! note "Backup recommended" - The key is stored in `~/.config/flowfile/`. Back up this directory to preserve access to your encrypted secrets. +**Per-user derivation.** Each user's secrets are encrypted with a key *derived* from the master key plus that user's ID (HKDF-SHA256). User A's derived key cannot decrypt User B's secrets, even though both come from the same master key. This isolates accidental cross-user reads at the cryptographic level. -### Docker +**Envelope format.** Each encrypted value in the database is stored as `$ffsec$1$$`. The `1` is a version digit reserved for future format changes. The `` lets the worker process decrypt without being separately told who the secret belongs to. + +## Where the master key lives + +| Mode | Location | +|------|----------| +| **Desktop (Electron)** | Auto-generated on first launch. Files: `~/.config/flowfile/.secret_key` (store-encryption key) + `~/.config/flowfile/flowfile.json.enc` (encrypted master key). On Windows: `%APPDATA%\flowfile\`. | +| **Python API** | Same paths as Desktop. Auto-generated on first call to anything that needs encryption. | +| **Docker** | Supplied externally via `FLOWFILE_MASTER_KEY` environment variable, or via the Docker secret mounted at `/run/secrets/flowfile_master_key`. Never written to disk inside the container. | + +### Desktop & Python API: first-run setup + +The first time Flowfile needs a master key, it generates one and prints a one-time warning to your logs: + +``` +FLOWFILE MASTER KEY GENERATED — BACK UP THESE FILES IMMEDIATELY. + fingerprint: 37028446 + file 1 (store key): /home/you/.config/flowfile/.secret_key + file 2 (encrypted secrets store): /home/you/.config/flowfile/flowfile.json.enc +If either file is lost, every secret encrypted by this Flowfile instance +becomes permanently unrecoverable. There is no recovery path. +``` + +You will see this exactly once — on the very first run. Subsequent runs use the existing key silently. Copy the fingerprint somewhere safe; it's how you'll verify a backup later. + +### Docker: setup wizard On first start without a master key, Flowfile shows a setup screen: 1. Click **Generate Master Key** -2. Copy the generated key +2. Copy the generated key **and** the displayed fingerprint 3. Add to your `.env` file: `FLOWFILE_MASTER_KEY=` 4. Restart the containers ![Setup Wizard](../../../assets/images/guides/docker-deployment/setup_wizard.png) -!!! danger "Protect your master key" - - Back up your `.env` file securely - - Never commit to version control - - Losing it = losing access to all encrypted secrets +## Backing up the master key + +**Desktop & Python API:** copy both files together (they're paired — neither is useful without the other): + +```bash +cp ~/.config/flowfile/.secret_key ~/.config/flowfile/flowfile.json.enc /your/backup/location/ +``` + +**Docker:** the key is the value of `FLOWFILE_MASTER_KEY` in your `.env`. Back up `.env`. Never commit it to version control. + +### Verifying a backup matches the live key -## Creating Secrets +The 8-character fingerprint printed at generation is `SHA-256(key)[:8]`. It's a one-way derivation, so it's safe to write down or share with a teammate — it can't be reversed back to the key. If you ever doubt whether a backup file matches the running instance, compare fingerprints. + +You can re-derive the fingerprint from Python: + +```python +import hashlib +key = "" +print(hashlib.sha256(key.encode()).hexdigest()[:8]) +``` + +!!! note "Pair the fingerprint with the date you generated the key" + A bare fingerprint is useless six months later if you've forgotten which install it belongs to. Note it as `flowfile master key fingerprint generated `. + +## Creating secrets 1. Open the **Connections** page from the left sidebar and select the **Secrets** tab 2. Click **Add Secret** @@ -49,15 +93,90 @@ On first start without a master key, Flowfile shows a setup screen: 4. Enter value 5. Save - ![Secrets Panel](../../../assets/images/guides/secrets/secrets_panel.png) -## Using Secrets +## Using secrets + +Reference secrets by name when configuring connections. The encrypted value is decrypted at runtime; the plaintext never appears in flow files or logs. + +## Audit log + +Every secret create / list / read / delete attempt — successful or failed — is recorded in the `secret_access_events` table. Each row captures: + +- The acting user ID +- The action (`create`, `list`, `read`, `delete`) +- The secret name (when applicable) +- Result status (`success` or `error`) and error code (e.g. `not_found`, `duplicate_name`) +- Source IP address (best-effort, honors `X-Forwarded-For`) +- Timestamp + +Admins can query the log via `GET /secrets/secrets/audit` (with optional `secret_name`, `action`, and `limit` query parameters). Non-admins see only their own events. + +**What is *not* recorded:** per-secret decrypt events during flow execution. Logging every decrypt would dominate the table with routine activity. If you need decrypt-time accounting later, that's a separate feature. + +## When things go wrong + +Three real scenarios, in priority order. Identify which one you're in *before* you act — the wrong response can make a recoverable situation unrecoverable, and the right response for a real breach doesn't involve the Flowfile UI at all. + +### 1. You lost the master key + +Either `.secret_key` or `flowfile.json.enc` is gone (or the `FLOWFILE_MASTER_KEY` value is forgotten). **There is no recovery path.** Every encrypted secret in your database is now plaintext-unrecoverable garbage. + +What to do: + +1. Delete the `secrets` table content (it's just unusable bytes now) and the connection rows that depend on it. +2. Re-enter every credential by hand — open each connection in the UI and paste the password again. +3. Set up a backup procedure before this happens again. See [Backing up the master key](#backing-up-the-master-key). + +!!! danger "There is no Flowfile-side recovery." + The encryption is intentionally one-way without the key. Don't open a support issue expecting a backdoor — there isn't one, and there shouldn't be one. + +### 2. The master key leaked, but you are confident the secrets database did not + +Example: a developer accidentally committed `.env` to a public repo for ten minutes; the production database is on a separate hardened host with audit logs showing no unauthorized access. + +Here, rotating the master key alone is a real mitigation. The attacker has the key but no ciphertext, so they can't decrypt anything offline. Once you swap the live key, even if they later breach the database, the stolen key is useless. + +What to do: + +1. Generate a new master key. +2. Re-enter every credential into Flowfile under the new key. (Flowfile does not currently offer transparent in-place rotation; the deferred feature would re-encrypt rows automatically. For now, re-enter manually.) +3. Retire the old key — delete the leaked copy from any backup that may have shared the exposure. +4. Verify the new fingerprint and store it in your backup record. + +### 3. The master key leaked AND the secrets database may have leaked too + +This is the conservative default whenever you can't *prove* scenario 2. In practice the master key file and the database often share the same exposure surface — same host, same backup, same compromised account, same git history. + +In this scenario, **rotating the Flowfile master key does almost nothing.** The attacker has already decrypted the dumped `secrets` table offline on their own machine. They walk away with plaintext database passwords, plaintext cloud credentials, plaintext API tokens. Those plaintext credentials are what's dangerous, and they're permanently outside your control. + +What to do — *in this order*: + +1. **Rotate every upstream credential at its source.** Change every database password (on the database itself), regenerate every cloud access key (in IAM / Azure AD / GCP IAM), revoke every API token (in the issuer's dashboard). This is the load-bearing step — it makes the attacker's plaintext copies useless. +2. **Notify anyone affected** — coworkers, customers depending on those services, your security team, your auditors. +3. **Generate a new Flowfile master key** and re-enter the *new* (rotated-at-source) credentials. The old encrypted database is now defunct; clear it. +4. Then continue with the operational hygiene checklist below. + +!!! danger "The master key alone is not the mitigation." + Re-encrypting Flowfile's database with a fresh key doesn't reach into your DB / cloud provider / API issuer to change anything there. Only rotating at the source does. Treat steps 1–2 as the actual fix; step 3 is cleanup. + +## Operational recommendations + +A short checklist to keep the next incident smaller than the last one. + +- [ ] Back up `.secret_key` + `flowfile.json.enc` (or `.env` for Docker) the day you set up Flowfile. +- [ ] Store that backup on **different media** from your database backup. Co-located backups defeat the point of separation. +- [ ] Write down the 8-char fingerprint together with the generation date. Stick it in your password manager or secrets vault. +- [ ] Use Docker mode (or env-var injection) for any deployment beyond local single-user — it keeps the key out of the database server's filesystem. +- [ ] Make sure `.env`, `master_key.txt`, and `*.json.enc` are in `.gitignore`. Never commit them. +- [ ] Periodically read the audit log (`GET /secrets/secrets/audit`) for read/delete activity you can't explain. + +## Encryption details -Reference secrets by name when configuring connections. The encrypted value is decrypted at runtime. +For the curious: -## Encryption +- **Algorithm:** Fernet — AES-128-CBC for confidentiality, HMAC-SHA256 for authentication, both keys derived from a single 256-bit input. +- **Key derivation:** HKDF-SHA256 with `salt="flowfile-secrets-v1"` and `info=f"user-{user_id}"`. Deterministic, so the same master + user always derives the same Fernet key. +- **Storage:** encrypted value lives in the `encrypted_value` column of the `secrets` table in your Flowfile catalog DB. -- **Algorithm**: Fernet (AES-128-CBC + HMAC-SHA256) -- **Isolation**: Each user's secrets stored separately -- **Storage**: Encrypted in SQLite database +For more on the internals — envelope versions, the shared crypto module, the audit table schema — see the developer reference at `docs/for-developers/secret-storage.md`. diff --git a/flowfile_core/flowfile_core/alembic/versions/012_secret_access_events.py b/flowfile_core/flowfile_core/alembic/versions/012_secret_access_events.py new file mode 100644 index 000000000..29215bb52 --- /dev/null +++ b/flowfile_core/flowfile_core/alembic/versions/012_secret_access_events.py @@ -0,0 +1,64 @@ +"""Add secret_access_events table for secret CRUD audit logging. + +This table captures every attempt to create, list, read, or delete a secret via +the public secrets API. The goal is operator-visible forensics — "who touched +what, when, and from where" — independent of the per-decrypt activity that +happens deep inside flow execution. Decrypt-on-flow events are intentionally +not logged here to keep the audit table at a useful signal-to-noise ratio. + +``secret_id`` is FK-with-SET-NULL on the existing ``secrets`` table so a row +survives a secret deletion (otherwise we'd lose the very record we need to +investigate the deletion). ``secret_name`` is denormalized into the audit row +for the same reason. + +Revision ID: 012 +Revises: 011 +Create Date: 2026-05-12 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "012" +down_revision: str | None = "011" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "secret_access_events", + sa.Column("id", sa.Integer, primary_key=True, index=True), + sa.Column( + "user_id", + sa.Integer, + sa.ForeignKey("users.id"), + nullable=False, + index=True, + ), + sa.Column( + "secret_id", + sa.Integer, + sa.ForeignKey("secrets.id", ondelete="SET NULL"), + nullable=True, + ), + sa.Column("secret_name", sa.String, nullable=True, index=True), + sa.Column("action", sa.String, nullable=False, index=True), + sa.Column("result_status", sa.String, nullable=False), + sa.Column("error", sa.Text, nullable=True), + sa.Column("source", sa.String, nullable=False, server_default="api"), + sa.Column("ip_address", sa.String, nullable=True), + sa.Column( + "created_at", + sa.DateTime, + server_default=sa.func.now(), + nullable=False, + index=True, + ), + ) + + +def downgrade() -> None: + op.drop_table("secret_access_events") diff --git a/flowfile_core/flowfile_core/auth/secrets.py b/flowfile_core/flowfile_core/auth/secrets.py index dc1ba6dfb..5739effef 100644 --- a/flowfile_core/flowfile_core/auth/secrets.py +++ b/flowfile_core/flowfile_core/auth/secrets.py @@ -2,6 +2,7 @@ Secure storage module for FlowFile credentials and secrets. """ +import hashlib import json import logging import os @@ -12,6 +13,16 @@ logger = logging.getLogger(__name__) +def _key_fingerprint(key: str) -> str: + """Short, stable, non-reversible identifier for a Fernet key. + + The first 8 hex chars of SHA-256 are enough for an operator to + visually compare a backed-up key against the live one without + exposing the raw key material in logs or status responses. + """ + return hashlib.sha256(key.encode()).hexdigest()[:8] + + class SecureStorage: """A secure local storage mechanism for secrets using Fernet encryption.""" @@ -220,4 +231,20 @@ def get_master_key(): if not key: key = Fernet.generate_key().decode() set_password("flowfile", "master_key", key) + # Loud one-shot banner. This branch only fires when the secure store + # had no master key — i.e. the *very first* call on a fresh install. + # The absence of the key file is itself the trigger, so no sentinel is + # needed. Subsequent runs read the cached key and skip this entirely. + storage_dir = _storage.storage_path + logger.warning( + "FLOWFILE MASTER KEY GENERATED — BACK UP THESE FILES IMMEDIATELY.\n" + " fingerprint: %s\n" + " file 1 (store key): %s\n" + " file 2 (encrypted secrets store): %s\n" + "If either file is lost, every secret encrypted by this Flowfile " + "instance becomes permanently unrecoverable. There is no recovery path.", + _key_fingerprint(key), + storage_dir / ".secret_key", + storage_dir / "flowfile.json.enc", + ) return key diff --git a/flowfile_core/flowfile_core/database/models.py b/flowfile_core/flowfile_core/database/models.py index 487432182..2a5556179 100644 --- a/flowfile_core/flowfile_core/database/models.py +++ b/flowfile_core/flowfile_core/database/models.py @@ -594,3 +594,37 @@ class AiProviderCredential(Base): api_key_secret = relationship("Secret", foreign_keys=[api_key_secret_id], lazy="joined") __table_args__ = (UniqueConstraint("user_id", "provider", name="uq_ai_provider_per_user"),) + + +# ==================== Secret Access Audit ==================== + + +class SecretAccessEvent(Base): + """One row per attempted access to a user's secret. + + Captures CRUD activity routed through the secrets API (create/list/read/ + delete). The intent is operator-visible forensics ("who touched what, + when, from where") rather than per-decrypt accounting — deep-stack decrypt + calls during flow execution are deliberately *not* logged here to keep the + audit table from being dominated by routine flow runs. + + ``secret_id`` is FK-with-SET-NULL so deleting a secret retains the audit + trail; ``secret_name`` is duplicated as plain text for the same reason. + """ + + __tablename__ = "secret_access_events" + + id = Column(Integer, primary_key=True, index=True) + user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True) + secret_id = Column( + Integer, + ForeignKey("secrets.id", ondelete="SET NULL"), + nullable=True, + ) + secret_name = Column(String, nullable=True, index=True) + action = Column(String, nullable=False, index=True) # 'create' | 'list' | 'read' | 'delete' + result_status = Column(String, nullable=False) # 'success' | 'error' + error = Column(Text, nullable=True) + source = Column(String, nullable=False, default="api", server_default="api") + ip_address = Column(String, nullable=True) + created_at = Column(DateTime, default=func.now(), nullable=False, index=True) diff --git a/flowfile_core/flowfile_core/routes/public.py b/flowfile_core/flowfile_core/routes/public.py index 55e96af5e..ad009bd22 100644 --- a/flowfile_core/flowfile_core/routes/public.py +++ b/flowfile_core/flowfile_core/routes/public.py @@ -4,7 +4,7 @@ from fastapi.responses import RedirectResponse from pydantic import BaseModel -from flowfile_core.auth.secrets import generate_master_key, is_master_key_configured +from flowfile_core.auth.secrets import _key_fingerprint, generate_master_key, is_master_key_configured router = APIRouter() @@ -21,6 +21,7 @@ class GeneratedKey(BaseModel): """Response model for the generate key endpoint.""" key: str + fingerprint: str instructions: str @@ -46,8 +47,13 @@ async def get_setup_status(): async def generate_key(): """Generate a new master encryption key.""" key = generate_master_key() + fingerprint = _key_fingerprint(key) instructions = ( + "BACK UP THIS KEY. If you lose it, every secret encrypted by this " + "Flowfile instance becomes permanently unrecoverable — there is no " + "recovery path.\n\n" + f"Key fingerprint (for verifying backups): {fingerprint}\n\n" f'Add to your .env file:\n FLOWFILE_MASTER_KEY="{key}"\n\n' "Then restart: docker-compose down && docker-compose up" ) - return GeneratedKey(key=key, instructions=instructions) + return GeneratedKey(key=key, fingerprint=fingerprint, instructions=instructions) diff --git a/flowfile_core/flowfile_core/routes/secrets.py b/flowfile_core/flowfile_core/routes/secrets.py index 2a2cb9312..707c7502d 100644 --- a/flowfile_core/flowfile_core/routes/secrets.py +++ b/flowfile_core/flowfile_core/routes/secrets.py @@ -4,31 +4,77 @@ This router provides secure endpoints for creating, retrieving, and deleting sensitive credentials for the authenticated user. Secrets are encrypted before being stored and are associated with the user's ID. + +Every CRUD endpoint emits a row to ``secret_access_events`` so an operator can +later reconstruct who touched what, when, and from where — see +:mod:`flowfile_core.secret_manager.audit`. Failed attempts are recorded too, +which is the entire point of the audit trail. """ import os +from datetime import datetime -from fastapi import APIRouter, Depends, HTTPException +from fastapi import APIRouter, Depends, HTTPException, Query, Request +from pydantic import BaseModel from sqlalchemy.orm import Session from flowfile_core.auth.jwt import get_current_active_user from flowfile_core.auth.models import Secret, SecretInput from flowfile_core.database import models as db_models from flowfile_core.database.connection import get_db +from flowfile_core.secret_manager import audit from flowfile_core.secret_manager.secret_manager import delete_secret as delete_secret_action from flowfile_core.secret_manager.secret_manager import store_secret router = APIRouter(dependencies=[Depends(get_current_active_user)]) +def _client_ip(request: Request) -> str | None: + """Best-effort client IP extraction. + + Prefers the forwarded-for header when the deployment sits behind a proxy, + falls back to the direct connection. Returned as a string or ``None`` if + nothing is available (e.g. in some test clients). + """ + forwarded = request.headers.get("x-forwarded-for") + if forwarded: + return forwarded.split(",")[0].strip() + client = request.client + return client.host if client else None + + +class SecretAuditEntry(BaseModel): + """Response shape for a single audit row. + + Mirrors the ORM columns minus the FK link — clients see the denormalized + ``secret_name`` and ``secret_id`` directly without a join. + """ + + id: int + user_id: int + secret_id: int | None + secret_name: str | None + action: str + result_status: str + error: str | None + source: str + ip_address: str | None + created_at: datetime + + @router.get("/secrets", response_model=list[Secret]) -async def get_secrets(current_user=Depends(get_current_active_user), db: Session = Depends(get_db)): +async def get_secrets( + request: Request, + current_user=Depends(get_current_active_user), + db: Session = Depends(get_db), +): """Retrieves all secret names for the currently authenticated user. Note: This endpoint returns the secret names and metadata but does not expose the decrypted secret values. Args: + request: The HTTP request (used to capture client IP for audit). current_user: The authenticated user object, injected by FastAPI. db: The database session, injected by FastAPI. @@ -37,39 +83,40 @@ async def get_secrets(current_user=Depends(get_current_active_user), db: Session """ user_id = current_user.id - # Get secrets from database db_secrets = db.query(db_models.Secret).filter(db_models.Secret.user_id == user_id).all() - - # Prepare response model (without decrypting) - secrets = [] - for db_secret in db_secrets: - secrets.append(Secret(name=db_secret.name, value=db_secret.encrypted_value, user_id=str(db_secret.user_id))) + secrets = [ + Secret(name=db_secret.name, value=db_secret.encrypted_value, user_id=str(db_secret.user_id)) + for db_secret in db_secrets + ] + + audit.record_event( + audit.SecretEvent( + user_id=user_id, + action="list", + result_status="success", + ip_address=_client_ip(request), + ), + db=db, + ) + db.commit() return secrets @router.post("/secrets", response_model=Secret) async def create_secret( - secret: SecretInput, current_user=Depends(get_current_active_user), db: Session = Depends(get_db) + secret: SecretInput, + request: Request, + current_user=Depends(get_current_active_user), + db: Session = Depends(get_db), ) -> Secret: """Creates a new secret for the authenticated user. The secret value is encrypted before being stored in the database. A secret name must be unique for a given user. - - Args: - secret: A `SecretInput` object containing the name and plaintext value of the secret. - current_user: The authenticated user object, injected by FastAPI. - db: The database session, injected by FastAPI. - - Raises: - HTTPException: 400 if a secret with the same name already exists for the user. - - Returns: - A `Secret` object containing the name and the *encrypted* value. """ - # Get user ID user_id = 1 if os.environ.get("FLOWFILE_MODE") == "electron" else current_user.id + ip = _client_ip(request) existing_secret = ( db.query(db_models.Secret) @@ -78,37 +125,93 @@ async def create_secret( ) if existing_secret: + audit.record_event( + audit.SecretEvent( + user_id=user_id, + action="create", + result_status="error", + secret_name=secret.name, + error="duplicate_name", + ip_address=ip, + ), + db=db, + ) + db.commit() raise HTTPException(status_code=400, detail="Secret with this name already exists") - # The store_secret function handles encryption and DB storage stored_secret = store_secret(db, secret, user_id) + + audit.record_event( + audit.SecretEvent( + user_id=user_id, + action="create", + result_status="success", + secret_name=stored_secret.name, + secret_id=stored_secret.id, + ip_address=ip, + ), + db=db, + ) + db.commit() + return Secret(name=stored_secret.name, value=stored_secret.encrypted_value, user_id=str(user_id)) +@router.get("/secrets/audit", response_model=list[SecretAuditEntry]) +async def get_audit_log( + request: Request, + secret_name: str | None = Query(None, description="Filter by secret name (exact match)"), + action: str | None = Query(None, description="Filter by action: create | list | read | delete"), + limit: int = Query(100, ge=1, le=1000, description="Max rows returned (DESC by created_at)"), + current_user=Depends(get_current_active_user), + db: Session = Depends(get_db), +) -> list[SecretAuditEntry]: + """Read the secret-access audit log. + + Non-admin users see only their own events; admins see every user's events + (operators investigating an incident shouldn't have to swap accounts). + """ + scope_user = None if current_user.is_admin else current_user.id + + rows = audit.query_events( + user_id=scope_user, + secret_name=secret_name, + action=action, # type: ignore[arg-type] + limit=limit, + db=db, + ) + return [ + SecretAuditEntry( + id=row.id, + user_id=row.user_id, + secret_id=row.secret_id, + secret_name=row.secret_name, + action=row.action, + result_status=row.result_status, + error=row.error, + source=row.source, + ip_address=row.ip_address, + created_at=row.created_at, + ) + for row in rows + ] + + @router.get("/secrets/{secret_name}", response_model=Secret) async def get_secret( - secret_name: str, current_user=Depends(get_current_active_user), db: Session = Depends(get_db) + secret_name: str, + request: Request, + current_user=Depends(get_current_active_user), + db: Session = Depends(get_db), ) -> Secret: """Retrieves a specific secret by name for the authenticated user. Note: This endpoint returns the secret name and metadata but does not expose the decrypted secret value. - - Args: - secret_name: The name of the secret to retrieve. - current_user: The authenticated user object, injected by FastAPI. - db: The database session, injected by FastAPI. - - Raises: - HTTPException: 404 if the secret is not found. - - Returns: - A `Secret` object containing the name and encrypted value. """ - # Get user ID user_id = 1 if os.environ.get("FLOWFILE_MODE") == "electron" else current_user.id + ip = _client_ip(request) - # Get secret from database db_secret = ( db.query(db_models.Secret) .filter(db_models.Secret.user_id == user_id, db_models.Secret.name == secret_name) @@ -116,26 +219,86 @@ async def get_secret( ) if not db_secret: + audit.record_event( + audit.SecretEvent( + user_id=user_id, + action="read", + result_status="error", + secret_name=secret_name, + error="not_found", + ip_address=ip, + ), + db=db, + ) + db.commit() raise HTTPException(status_code=404, detail="Secret not found") + audit.record_event( + audit.SecretEvent( + user_id=user_id, + action="read", + result_status="success", + secret_name=db_secret.name, + secret_id=db_secret.id, + ip_address=ip, + ), + db=db, + ) + db.commit() + return Secret(name=db_secret.name, value=db_secret.encrypted_value, user_id=str(db_secret.user_id)) @router.delete("/secrets/{secret_name}", status_code=204) async def delete_secret( - secret_name: str, current_user=Depends(get_current_active_user), db: Session = Depends(get_db) + secret_name: str, + request: Request, + current_user=Depends(get_current_active_user), + db: Session = Depends(get_db), ) -> None: - """Deletes a secret by name for the authenticated user. + """Deletes a secret by name for the authenticated user.""" + user_id = 1 if os.environ.get("FLOWFILE_MODE") == "electron" else current_user.id + ip = _client_ip(request) - Args: - secret_name: The name of the secret to delete. - current_user: The authenticated user object, injected by FastAPI. - db: The database session, injected by FastAPI. + # Look up before the delete so we can record the secret_id in the audit row; + # the FK in ``secret_access_events`` is SET NULL on delete, but capturing + # the id at the moment of the action makes joins against deleted rows easier. + target = ( + db.query(db_models.Secret) + .filter(db_models.Secret.user_id == user_id, db_models.Secret.name == secret_name) + .first() + ) + + try: + delete_secret_action(db, secret_name, user_id) + except HTTPException as exc: + # Map the 404 to a short error code so audit consumers can filter on it. + err = "not_found" if exc.status_code == 404 else str(exc.detail) + audit.record_event( + audit.SecretEvent( + user_id=user_id, + action="delete", + result_status="error", + secret_name=secret_name, + error=err, + ip_address=ip, + ), + db=db, + ) + db.commit() + raise + + audit.record_event( + audit.SecretEvent( + user_id=user_id, + action="delete", + result_status="success", + secret_name=secret_name, + secret_id=target.id if target else None, + ip_address=ip, + ), + db=db, + ) + db.commit() - Returns: - An empty response with a 204 No Content status code upon success. - """ - # Get user ID - user_id = 1 if os.environ.get("FLOWFILE_MODE") == "electron" else current_user.id - delete_secret_action(db, secret_name, user_id) return None diff --git a/flowfile_core/flowfile_core/secret_manager/audit.py b/flowfile_core/flowfile_core/secret_manager/audit.py new file mode 100644 index 000000000..7b52b9b97 --- /dev/null +++ b/flowfile_core/flowfile_core/secret_manager/audit.py @@ -0,0 +1,140 @@ +"""Secret-access audit log — persistence + query layer. + +Every secret CRUD endpoint records exactly one row via :func:`record_event` so +operators can answer "who touched what, when, from where" after the fact. Read +access is via :func:`query_events`, exposed through the audit route. + +Decrypt activity during flow execution is *deliberately* not recorded here: +that path runs deep inside the engine where threading a DB session and user +context would be invasive, and the resulting volume would drown the genuinely +interesting CRUD events. If decrypt-time accounting becomes load-bearing later, +add a separate event source rather than expanding this one. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Literal + +from sqlalchemy.orm import Session + +from flowfile_core.database.connection import SessionLocal +from flowfile_core.database.models import SecretAccessEvent + +logger = logging.getLogger(__name__) + +Action = Literal["create", "list", "read", "update", "delete"] +ResultStatus = Literal["success", "error"] + + +@dataclass(slots=True) +class SecretEvent: + """In-memory representation of a secret-access attempt before persistence. + + ``secret_name`` is optional because ``list`` doesn't target a specific name. + ``secret_id`` is optional because pre-create the row doesn't exist yet and + post-delete it has already been removed. + """ + + user_id: int + action: Action + result_status: ResultStatus + secret_name: str | None = None + secret_id: int | None = None + error: str | None = None + source: str = "api" + ip_address: str | None = None + + +def record_event(event: SecretEvent, db: Session | None = None) -> SecretAccessEvent: + """Persist a :class:`SecretEvent` to ``secret_access_events``. + + Pass ``db`` to keep the write inside the caller's transaction; pass ``None`` + to use a short-lived session that commits before returning. Audit writes + are best-effort: an exception here is logged and swallowed rather than + failing the user-facing request, because a CRUD operation that already + succeeded shouldn't roll back over a missing audit row. + """ + row = SecretAccessEvent( + user_id=event.user_id, + secret_id=event.secret_id, + secret_name=event.secret_name, + action=event.action, + result_status=event.result_status, + error=event.error, + source=event.source, + ip_address=event.ip_address, + ) + + try: + if db is not None: + db.add(row) + db.flush() + else: + with SessionLocal() as session: + session.add(row) + session.commit() + session.refresh(row) + session.expunge(row) + except Exception as exc: + logger.warning( + "secret_audit_write_failed user=%s action=%s name=%s err=%s", + event.user_id, + event.action, + event.secret_name, + exc, + ) + return row + + logger.info( + "secret_audit user=%s action=%s name=%s status=%s source=%s", + event.user_id, + event.action, + event.secret_name or "-", + event.result_status, + event.source, + ) + return row + + +def query_events( + *, + user_id: int | None = None, + secret_name: str | None = None, + action: Action | None = None, + limit: int = 100, + db: Session | None = None, +) -> list[SecretAccessEvent]: + """Read audit events. Filters compose; rows ordered DESC by ``created_at``.""" + + def _query(session: Session) -> list[SecretAccessEvent]: + q = session.query(SecretAccessEvent) + if user_id is not None: + q = q.filter(SecretAccessEvent.user_id == user_id) + if secret_name is not None: + q = q.filter(SecretAccessEvent.secret_name == secret_name) + if action is not None: + q = q.filter(SecretAccessEvent.action == action) + rows = ( + q.order_by(SecretAccessEvent.created_at.desc(), SecretAccessEvent.id.desc()) + .limit(limit) + .all() + ) + for row in rows: + session.expunge(row) + return rows + + if db is not None: + return _query(db) + with SessionLocal() as session: + return _query(session) + + +__all__ = [ + "Action", + "ResultStatus", + "SecretEvent", + "query_events", + "record_event", +] diff --git a/flowfile_core/flowfile_core/secret_manager/secret_manager.py b/flowfile_core/flowfile_core/secret_manager/secret_manager.py index 1ccf94aa0..f043d824e 100644 --- a/flowfile_core/flowfile_core/secret_manager/secret_manager.py +++ b/flowfile_core/flowfile_core/secret_manager/secret_manager.py @@ -1,8 +1,4 @@ -import base64 - from cryptography.fernet import Fernet -from cryptography.hazmat.primitives import hashes -from cryptography.hazmat.primitives.kdf.hkdf import HKDF from fastapi.exceptions import HTTPException from pydantic import SecretStr from sqlalchemy import and_ @@ -12,141 +8,86 @@ from flowfile_core.auth.secrets import get_master_key from flowfile_core.database import models as db_models from flowfile_core.database.connection import get_db_context - -# Version identifier for key derivation scheme (allows future migrations) -KEY_DERIVATION_VERSION = b"flowfile-secrets-v1" - -# Encrypted secret format: $ffsec$1${user_id}${fernet_token} -# This embeds the user_id so the worker can derive the correct key -SECRET_FORMAT_PREFIX = "$ffsec$1$" +from shared.crypto.envelope import ( + KEY_DERIVATION_VERSION, + SECRET_FORMAT_PREFIX, + decrypt_secret_envelope, + encrypt_secret_envelope, +) +from shared.crypto.envelope import ( + derive_user_key as _derive_user_key, +) + +__all__ = [ + "KEY_DERIVATION_VERSION", + "SECRET_FORMAT_PREFIX", + "SecretInput", + "_decrypt_with_master_key", + "_encrypt_with_master_key", + "decrypt_secret", + "delete_secret", + "derive_user_key", + "encrypt_secret", + "get_encrypted_secret", + "store_secret", +] def derive_user_key(user_id: int) -> bytes: - """ - Derive a user-specific encryption key from the master key using HKDF. + """Derive a per-user Fernet key from the current master key.""" + return _derive_user_key(get_master_key().encode(), user_id) - This provides cryptographic isolation between users - each user's secrets - are encrypted with a unique key derived from the master key. - Args: - user_id: The unique identifier for the user +def _encrypt_with_master_key(secret_value: str) -> str: + """Legacy encryption with the master key directly (no per-user derivation). - Returns: - bytes: A 32-byte URL-safe base64-encoded key suitable for Fernet + Still used by older callers (e.g. ``flow_graph._encrypt_with_master_key``). + New code should use :func:`encrypt_secret` so the envelope embeds user_id. """ - master_key = get_master_key().encode() - - # Use HKDF to derive a user-specific key - hkdf = HKDF( - algorithm=hashes.SHA256(), - length=32, # Fernet requires 32 bytes - salt=KEY_DERIVATION_VERSION, # Static salt is fine for key derivation - info=f"user-{user_id}".encode(), # User-specific context - ) - - # Derive raw key material and encode for Fernet - derived_key = hkdf.derive(master_key) - return base64.urlsafe_b64encode(derived_key) - - -def _encrypt_with_master_key(secret_value: str) -> str: - """Legacy encryption using master key directly (for backward compatibility).""" key = get_master_key().encode() - f = Fernet(key) - return f.encrypt(secret_value.encode()).decode() + return Fernet(key).encrypt(secret_value.encode()).decode() def _decrypt_with_master_key(encrypted_value: str) -> SecretStr: - """Legacy decryption using master key directly (for backward compatibility).""" + """Legacy counterpart to :func:`_encrypt_with_master_key`.""" key = get_master_key().encode() - f = Fernet(key) - return SecretStr(f.decrypt(encrypted_value.encode()).decode()) + return SecretStr(Fernet(key).decrypt(encrypted_value.encode()).decode()) def encrypt_secret(secret_value: str, user_id: int) -> str: - """ - Encrypt a secret value using a user-specific derived key. - - The encrypted format embeds the user_id so it can be decrypted - without knowing the user context (e.g., by the worker service). - - Format: $ffsec$1${user_id}${fernet_token} - - Args: - secret_value: The plaintext secret to encrypt - user_id: The user ID to derive the encryption key for - - Returns: - str: The encrypted value with embedded user_id - """ - key = derive_user_key(user_id) - f = Fernet(key) - fernet_token = f.encrypt(secret_value.encode()).decode() - return f"{SECRET_FORMAT_PREFIX}{user_id}${fernet_token}" + """Encrypt ``secret_value`` as a v1 envelope bound to ``user_id``.""" + return encrypt_secret_envelope(get_master_key().encode(), secret_value, user_id) def decrypt_secret(encrypted_value: str, user_id: int | None = None) -> SecretStr: - """ - Decrypt an encrypted value. - - Supports both new format (with embedded user_id) and legacy format. - - New format: $ffsec$1${user_id}${fernet_token} - user_id extracted automatically - - Legacy format: raw Fernet token - requires user_id parameter or uses master key + """Decrypt a v1 envelope or a legacy raw token. - Args: - encrypted_value: The encrypted secret - user_id: Optional user ID (required for legacy secrets, ignored for new format) - - Returns: - SecretStr: The decrypted value wrapped in SecretStr + For v1 envelopes, ``user_id`` is read from the envelope. For legacy raw + tokens, the caller-supplied ``user_id`` selects the derived key; ``None`` + falls back to master-key decryption. """ - # Check for new versioned format with embedded user_id - if encrypted_value.startswith(SECRET_FORMAT_PREFIX): - # Parse: $ffsec$1${user_id}${fernet_token} - remainder = encrypted_value[len(SECRET_FORMAT_PREFIX) :] - parts = remainder.split("$", 1) - if len(parts) != 2: - raise ValueError("Invalid encrypted secret format") - - embedded_user_id = int(parts[0]) - fernet_token = parts[1] - - key = derive_user_key(embedded_user_id) - f = Fernet(key) - return SecretStr(f.decrypt(fernet_token.encode()).decode()) - - # Legacy format - use provided user_id or fall back to master key - if user_id is not None: - key = derive_user_key(user_id) - f = Fernet(key) - return SecretStr(f.decrypt(encrypted_value.encode()).decode()) - - # Fall back to master key for legacy secrets without user context - return _decrypt_with_master_key(encrypted_value) + return decrypt_secret_envelope(get_master_key().encode(), encrypted_value, user_id) def get_encrypted_secret(current_user_id: int, secret_name: str) -> str | None: with get_db_context() as db: - user_id = current_user_id db_secret = ( db.query(db_models.Secret) - .filter(and_(db_models.Secret.user_id == user_id, db_models.Secret.name == secret_name)) + .filter(and_(db_models.Secret.user_id == current_user_id, db_models.Secret.name == secret_name)) .first() ) if db_secret: return db_secret.encrypted_value - else: - return None + return None def store_secret(db: Session, secret: SecretInput, user_id: int) -> db_models.Secret: encrypted_value = encrypt_secret(secret.value.get_secret_value(), user_id) - # Store in database db_secret = db_models.Secret( name=secret.name, encrypted_value=encrypted_value, - iv="", # Legacy field, not used with current encryption + iv="", user_id=user_id, ) db.add(db_secret) diff --git a/flowfile_core/tests/test_secret_audit.py b/flowfile_core/tests/test_secret_audit.py new file mode 100644 index 000000000..e2048c8ae --- /dev/null +++ b/flowfile_core/tests/test_secret_audit.py @@ -0,0 +1,159 @@ +"""End-to-end coverage for the secret-access audit trail. + +The audit table is written to as a side effect of the CRUD endpoints; these +tests treat that side effect as the contract. Each test exercises one route, +then queries ``GET /secrets/secrets/audit`` to confirm the expected row +appears with the right status. +""" + +import os +import sys +import uuid + +from fastapi.testclient import TestClient + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..")) + +from flowfile_core import main +from flowfile_core.database.connection import get_db_context +from flowfile_core.database.models import SecretAccessEvent + + +def _get_client() -> TestClient: + """Authenticated test client. Mirrors ``test_endpoints.get_test_client``.""" + with TestClient(main.app) as bootstrap: + token = bootstrap.post("/auth/token").json()["access_token"] + client = TestClient(main.app) + client.headers = {"Authorization": f"Bearer {token}"} + return client + + +def _audit_rows(action: str | None = None, secret_name: str | None = None): + """Direct DB query so the test doesn't rely on the audit endpoint itself.""" + with get_db_context() as db: + q = db.query(SecretAccessEvent) + if action is not None: + q = q.filter(SecretAccessEvent.action == action) + if secret_name is not None: + q = q.filter(SecretAccessEvent.secret_name == secret_name) + rows = q.order_by(SecretAccessEvent.id.desc()).all() + for row in rows: + db.expunge(row) + return rows + + +def test_create_secret_emits_success_audit_row(): + client = _get_client() + name = f"audit_test_{uuid.uuid4().hex[:8]}" + + try: + r = client.post("/secrets/secrets", json={"name": name, "value": "v"}) + assert r.status_code == 200, r.text + + rows = _audit_rows(action="create", secret_name=name) + assert len(rows) == 1 + assert rows[0].result_status == "success" + assert rows[0].secret_id is not None + assert rows[0].source == "api" + finally: + client.delete(f"/secrets/secrets/{name}") + + +def test_create_duplicate_secret_emits_error_audit_row(): + client = _get_client() + name = f"audit_dup_{uuid.uuid4().hex[:8]}" + + try: + client.post("/secrets/secrets", json={"name": name, "value": "v"}) + r = client.post("/secrets/secrets", json={"name": name, "value": "v"}) + assert r.status_code == 400 + + rows = _audit_rows(action="create", secret_name=name) + # one success row + one error row + assert len(rows) == 2 + statuses = {row.result_status for row in rows} + assert statuses == {"success", "error"} + err_row = next(r for r in rows if r.result_status == "error") + assert err_row.error == "duplicate_name" + finally: + client.delete(f"/secrets/secrets/{name}") + + +def test_list_secrets_emits_audit_row(): + client = _get_client() + pre = len(_audit_rows(action="list")) + + r = client.get("/secrets/secrets") + assert r.status_code == 200 + + post = len(_audit_rows(action="list")) + assert post == pre + 1 + + +def test_read_missing_secret_emits_error_audit_row(): + client = _get_client() + name = f"audit_missing_{uuid.uuid4().hex[:8]}" + + r = client.get(f"/secrets/secrets/{name}") + assert r.status_code == 404 + + rows = _audit_rows(action="read", secret_name=name) + assert len(rows) == 1 + assert rows[0].result_status == "error" + assert rows[0].error == "not_found" + + +def test_delete_secret_emits_success_audit_row(): + client = _get_client() + name = f"audit_del_{uuid.uuid4().hex[:8]}" + + client.post("/secrets/secrets", json={"name": name, "value": "v"}) + r = client.delete(f"/secrets/secrets/{name}") + assert r.status_code == 204 + + rows = _audit_rows(action="delete", secret_name=name) + assert len(rows) == 1 + assert rows[0].result_status == "success" + + +def test_audit_endpoint_returns_recent_events(): + client = _get_client() + name = f"audit_ep_{uuid.uuid4().hex[:8]}" + + try: + client.post("/secrets/secrets", json={"name": name, "value": "v"}) + + r = client.get( + "/secrets/secrets/audit", + params={"secret_name": name, "limit": 10}, + ) + assert r.status_code == 200, r.text + body = r.json() + assert len(body) >= 1 + assert body[0]["secret_name"] == name + assert body[0]["action"] == "create" + assert body[0]["result_status"] == "success" + assert "created_at" in body[0] + finally: + client.delete(f"/secrets/secrets/{name}") + + +def test_audit_endpoint_filters_by_action(): + client = _get_client() + name = f"audit_filter_{uuid.uuid4().hex[:8]}" + + try: + client.post("/secrets/secrets", json={"name": name, "value": "v"}) + client.delete(f"/secrets/secrets/{name}") + + r = client.get( + "/secrets/secrets/audit", + params={"secret_name": name, "action": "delete", "limit": 10}, + ) + assert r.status_code == 200 + body = r.json() + assert len(body) == 1 + assert body[0]["action"] == "delete" + finally: + # Best-effort cleanup if the test failed before delete. + client.delete(f"/secrets/secrets/{name}") diff --git a/flowfile_worker/flowfile_worker/secrets.py b/flowfile_worker/flowfile_worker/secrets.py index 48265e277..e8b3b42ff 100644 --- a/flowfile_worker/flowfile_worker/secrets.py +++ b/flowfile_worker/flowfile_worker/secrets.py @@ -1,28 +1,45 @@ """ Simplified secure storage module for FlowFile worker to read credentials and secrets. + +The crypto envelope itself lives in :mod:`shared.crypto.envelope` so the +worker and core agree byte-for-byte on encoding. This module owns the +worker-side master-key lookup (Docker secret / env / local file) and re-exports +the shared envelope functions for backward compatibility with existing callers +and tests. """ -import base64 import json import logging import os from pathlib import Path from cryptography.fernet import Fernet -from cryptography.hazmat.primitives import hashes -from cryptography.hazmat.primitives.kdf.hkdf import HKDF from pydantic import SecretStr from flowfile_worker.configs import TEST_MODE +from shared.crypto.envelope import ( + KEY_DERIVATION_VERSION, + SECRET_FORMAT_PREFIX, + decrypt_secret_envelope, + encrypt_secret_envelope, +) +from shared.crypto.envelope import ( + derive_user_key as _derive_user_key, +) -# Set up logging logger = logging.getLogger(__name__) -# Version identifier for key derivation scheme (must match flowfile_core) -KEY_DERIVATION_VERSION = b"flowfile-secrets-v1" - -# Encrypted secret format: $ffsec$1${user_id}${fernet_token} -SECRET_FORMAT_PREFIX = "$ffsec$1$" +__all__ = [ + "KEY_DERIVATION_VERSION", + "SECRET_FORMAT_PREFIX", + "SecureStorage", + "decrypt_secret", + "derive_user_key", + "encrypt_secret", + "get_docker_secret_key", + "get_master_key", + "get_password", +] class SecureStorage: @@ -63,6 +80,12 @@ def get_password(self, service_name, username): _storage = SecureStorage() +# Cache for the auto-generated test master key. We generate one per-process when +# TEST_MODE is on and no explicit key is supplied via env, so an encrypt/decrypt +# round-trip in a single test process stays consistent without committing a key +# to source control. +_TEST_MASTER_KEY: str | None = None + def get_password(service_name, username): """ @@ -88,10 +111,8 @@ def get_docker_secret_key() -> str | None: Raises: RuntimeError: If the secret file exists but cannot be read, or key is invalid. """ - # First, check for environment variable (allows runtime configuration) env_key = os.environ.get("FLOWFILE_MASTER_KEY") if env_key: - # Validate it's a proper Fernet key try: Fernet(env_key.encode()) return env_key @@ -99,20 +120,17 @@ def get_docker_secret_key() -> str | None: logger.error("FLOWFILE_MASTER_KEY environment variable is not a valid Fernet key") raise RuntimeError("FLOWFILE_MASTER_KEY is not a valid Fernet key") from None - # Then, check for Docker secret file secret_path = "/run/secrets/flowfile_master_key" if os.path.exists(secret_path): try: with open(secret_path) as f: key = f.read().strip() - # Validate the key Fernet(key.encode()) return key except Exception as e: logger.error(f"Failed to read or validate master key from Docker secret: {e}") raise RuntimeError("Failed to read master key from Docker secret") from e - # No key configured return None @@ -120,9 +138,13 @@ def get_master_key() -> str: """ Get the master encryption key. - If in TEST_MODE, returns a test key. - If running in Docker, retrieves the key from Docker secrets or environment. - Otherwise, retrieves the key from secure storage. + Resolution order: + 1. If ``TEST_MODE`` is on, prefer ``FLOWFILE_MASTER_KEY`` (so core+worker + integration tests can share a key); otherwise generate a fresh per-process + key so worker-only round-trip tests have a usable key without storing a + value in source. + 2. If running in Docker, read the key from the Docker secret or env var. + 3. Otherwise read the key from the local SecureStorage. Returns: str: The master encryption key @@ -131,11 +153,15 @@ def get_master_key() -> str: RuntimeError: If in Docker mode and no key is configured. ValueError: If the master key is not found in storage. """ - # First check for test mode if TEST_MODE: - return b"06t640eu3AG2FmglZS0n0zrEdqadoT7lYDwgSmKyxE4=".decode() + explicit = os.environ.get("FLOWFILE_MASTER_KEY") + if explicit: + return explicit + global _TEST_MASTER_KEY + if _TEST_MASTER_KEY is None: + _TEST_MASTER_KEY = Fernet.generate_key().decode() + return _TEST_MASTER_KEY - # Next check if running in Docker if os.environ.get("FLOWFILE_MODE") == "docker": key = get_docker_secret_key() if key is None: @@ -145,7 +171,6 @@ def get_master_key() -> str: ) return key - # Otherwise read from local storage key = get_password("flowfile", "master_key") if not key: raise ValueError("Master key not found in storage.") @@ -153,89 +178,23 @@ def get_master_key() -> str: def derive_user_key(user_id: int) -> bytes: - """ - Derive a user-specific encryption key from the master key using HKDF. - - This provides cryptographic isolation between users - each user's secrets - are encrypted with a unique key derived from the master key. - - Args: - user_id: The unique identifier for the user - - Returns: - bytes: A 32-byte URL-safe base64-encoded key suitable for Fernet - """ - master_key = get_master_key().encode() - - # Use HKDF to derive a user-specific key - hkdf = HKDF( - algorithm=hashes.SHA256(), - length=32, # Fernet requires 32 bytes - salt=KEY_DERIVATION_VERSION, # Static salt is fine for key derivation - info=f"user-{user_id}".encode(), # User-specific context - ) - - # Derive raw key material and encode for Fernet - derived_key = hkdf.derive(master_key) - return base64.urlsafe_b64encode(derived_key) + """Derive a per-user Fernet key from the current master key.""" + return _derive_user_key(get_master_key().encode(), user_id) def decrypt_secret(encrypted_value: str) -> SecretStr: - """ - Decrypt an encrypted value. - - Supports both new format (with embedded user_id) and legacy format. - - New format: $ffsec$1${user_id}${fernet_token} - user_id extracted automatically - - Legacy format: raw Fernet token - uses master key directly - - Args: - encrypted_value: The encrypted value as a string - - Returns: - SecretStr: The decrypted value as a SecretStr - """ - # Check for new versioned format with embedded user_id - if encrypted_value.startswith(SECRET_FORMAT_PREFIX): - # Parse: $ffsec$1${user_id}${fernet_token} - remainder = encrypted_value[len(SECRET_FORMAT_PREFIX) :] - parts = remainder.split("$", 1) - if len(parts) != 2: - raise ValueError("Invalid encrypted secret format") - - embedded_user_id = int(parts[0]) - fernet_token = parts[1] - - key = derive_user_key(embedded_user_id) - f = Fernet(key) - return SecretStr(f.decrypt(fernet_token.encode()).decode()) - - # Legacy format - use master key directly - key = get_master_key().encode() - f = Fernet(key) - return SecretStr(f.decrypt(encrypted_value.encode()).decode()) + """Decrypt a v1 envelope (``$ffsec$1$...``) or a legacy raw Fernet token.""" + return decrypt_secret_envelope(get_master_key().encode(), encrypted_value) def encrypt_secret(secret_value: str, user_id: int | None = None) -> str: - """ - Encrypt a secret value. + """Encrypt a secret value. - If user_id is provided, uses per-user key derivation with embedded user_id format. - Otherwise, uses legacy master key encryption (for backward compatibility in tests). - - Args: - secret_value: The secret value to encrypt - user_id: Optional user ID for per-user key derivation - - Returns: - str: The encrypted value as a string + With ``user_id`` set, emits a v1 envelope (``$ffsec$1$$``). + Without ``user_id``, emits a legacy raw Fernet token encrypted with the + master key — kept for backward compatibility with existing worker tests. """ + master_key = get_master_key().encode() if user_id is not None: - key = derive_user_key(user_id) - f = Fernet(key) - fernet_token = f.encrypt(secret_value.encode()).decode() - return f"{SECRET_FORMAT_PREFIX}{user_id}${fernet_token}" - - # Legacy format for backward compatibility - key = get_master_key().encode() - f = Fernet(key) - return f.encrypt(secret_value.encode()).decode() + return encrypt_secret_envelope(master_key, secret_value, user_id) + return Fernet(master_key).encrypt(secret_value.encode()).decode() diff --git a/mkdocs.yml b/mkdocs.yml index 1d7be22cf..45b4400a9 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -149,6 +149,7 @@ nav: - Core Internals: for-developers/flowfile-core.md - Internal API Reference: for-developers/python-api-reference.md - Kernel Architecture: for-developers/kernel-architecture.md + - Secret Storage: for-developers/secret-storage.md - AI Integration Architecture: for-developers/ai-architecture.md - Visualizations & Graphic Walker: for-developers/visualizations.md - Create Custom Nodes: for-developers/creating-custom-nodes.md diff --git a/shared/crypto/__init__.py b/shared/crypto/__init__.py new file mode 100644 index 000000000..2088f4882 --- /dev/null +++ b/shared/crypto/__init__.py @@ -0,0 +1,26 @@ +"""Shared cryptographic primitives for Flowfile secret storage. + +Both ``flowfile_core`` and ``flowfile_worker`` need to encrypt/decrypt secrets +using the same envelope format. This package owns the single source of truth +for that logic — services fetch the master key via their own backend +(``flowfile_core.auth.secrets`` or ``flowfile_worker.secrets``) and then call +into this module for the actual crypto. +""" + +from shared.crypto.envelope import ( + KEY_DERIVATION_VERSION, + SECRET_FORMAT_PREFIX, + decrypt_secret_envelope, + derive_user_key, + encrypt_secret_envelope, + parse_v1_envelope, +) + +__all__ = [ + "KEY_DERIVATION_VERSION", + "SECRET_FORMAT_PREFIX", + "decrypt_secret_envelope", + "derive_user_key", + "encrypt_secret_envelope", + "parse_v1_envelope", +] diff --git a/shared/crypto/envelope.py b/shared/crypto/envelope.py new file mode 100644 index 000000000..66d893246 --- /dev/null +++ b/shared/crypto/envelope.py @@ -0,0 +1,96 @@ +"""Cryptographic envelope for Flowfile secrets. + +Owns the single source of truth for how secrets are derived, encrypted, and +parsed across ``flowfile_core`` and ``flowfile_worker``. Both services fetch +the master key through their own backend, then delegate the actual crypto +operations to this module. + +Envelope format v1: ``$ffsec$1$$`` + +The user_id is embedded so a downstream consumer (e.g. the worker) can decrypt +without separately being told who the secret belongs to. The leading +``$ffsec$1$`` prefix reserves the version digit for forward-compatible envelope +changes — a future v2 envelope can be detected by the same dispatch. + +Changing ``KEY_DERIVATION_VERSION`` invalidates every existing secret because +it is the HKDF salt. Do not edit without a migration path. +""" + +import base64 + +from cryptography.fernet import Fernet +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.hkdf import HKDF +from pydantic import SecretStr + +KEY_DERIVATION_VERSION = b"flowfile-secrets-v1" + +SECRET_FORMAT_PREFIX = "$ffsec$1$" + + +def derive_user_key(master_key: bytes, user_id: int) -> bytes: + """Derive a per-user 32-byte Fernet key from the master key using HKDF. + + Deterministic: the same ``master_key`` + ``user_id`` always produces the + same key, which is what lets secrets remain decryptable across process + restarts without persisting per-user keys. + """ + hkdf = HKDF( + algorithm=hashes.SHA256(), + length=32, + salt=KEY_DERIVATION_VERSION, + info=f"user-{user_id}".encode(), + ) + derived = hkdf.derive(master_key) + return base64.urlsafe_b64encode(derived) + + +def parse_v1_envelope(encrypted_value: str) -> tuple[int, str]: + """Split a v1 envelope into ``(user_id, fernet_token)``. + + Raises ``ValueError`` if ``encrypted_value`` isn't a well-formed v1 envelope. + """ + if not encrypted_value.startswith(SECRET_FORMAT_PREFIX): + raise ValueError("Not a v1 envelope") + remainder = encrypted_value[len(SECRET_FORMAT_PREFIX) :] + parts = remainder.split("$", 1) + if len(parts) != 2: + raise ValueError("Invalid encrypted secret format") + try: + user_id = int(parts[0]) + except ValueError as e: + raise ValueError("Invalid encrypted secret format") from e + return user_id, parts[1] + + +def encrypt_secret_envelope(master_key: bytes, plaintext: str, user_id: int) -> str: + """Encrypt ``plaintext`` as a v1 envelope: ``$ffsec$1$$``.""" + key = derive_user_key(master_key, user_id) + token = Fernet(key).encrypt(plaintext.encode()).decode() + return f"{SECRET_FORMAT_PREFIX}{user_id}${token}" + + +def decrypt_secret_envelope( + master_key: bytes, + encrypted_value: str, + user_id: int | None = None, +) -> SecretStr: + """Decrypt a v1 envelope or a legacy raw Fernet token. + + For v1 envelopes (``$ffsec$1$...``), ``user_id`` is extracted from the blob + and the caller-supplied ``user_id`` argument is ignored. + + For legacy raw tokens (no envelope prefix), the optional ``user_id`` selects + a derived key; passing ``None`` falls back to decrypting with the master key + directly — which is how the earliest version of the secret store encrypted. + """ + if encrypted_value.startswith(SECRET_FORMAT_PREFIX): + embedded_user_id, token = parse_v1_envelope(encrypted_value) + key = derive_user_key(master_key, embedded_user_id) + return SecretStr(Fernet(key).decrypt(token.encode()).decode()) + + if user_id is not None: + key = derive_user_key(master_key, user_id) + return SecretStr(Fernet(key).decrypt(encrypted_value.encode()).decode()) + + return SecretStr(Fernet(master_key).decrypt(encrypted_value.encode()).decode()) From d83928710e41301c838e028ac6066ff5206800df Mon Sep 17 00:00:00 2001 From: edwardvaneechoud Date: Thu, 14 May 2026 10:11:14 +0200 Subject: [PATCH 2/3] ensure worker has same key access --- flowfile_worker/tests/conftest.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/flowfile_worker/tests/conftest.py b/flowfile_worker/tests/conftest.py index 682df0040..1f4bc054f 100644 --- a/flowfile_worker/tests/conftest.py +++ b/flowfile_worker/tests/conftest.py @@ -5,8 +5,10 @@ import sys import pytest +from cryptography.fernet import Fernet os.environ['TEST_MODE'] = '1' +os.environ.setdefault('FLOWFILE_MASTER_KEY', Fernet.generate_key().decode()) from tests.utils import is_docker_available From 294ca421c50320fa2fafe9e498bfedab4281cfc6 Mon Sep 17 00:00:00 2001 From: edwardvaneechoud Date: Tue, 26 May 2026 21:10:43 +0200 Subject: [PATCH 3/3] Fixing bugs with secrets --- docs/for-developers/secret-storage.md | 2 +- docs/users/visual-editor/catalog/secrets.md | 10 +++++-- ..._events.py => 015_secret_access_events.py} | 8 +++--- flowfile_core/flowfile_core/auth/secrets.py | 6 ++-- flowfile_core/flowfile_core/routes/secrets.py | 15 ++++++---- .../flowfile_core/secret_manager/audit.py | 4 +-- .../tests/test_master_key_normalization.py | 28 +++++++++++++++++++ flowfile_core/tests/test_secret_audit.py | 19 +++++++++++++ flowfile_worker/flowfile_worker/secrets.py | 4 ++- flowfile_worker/tests/test_secrets.py | 12 ++++++++ shared/crypto/__init__.py | 2 ++ shared/crypto/master_key.py | 18 ++++++++++++ 12 files changed, 110 insertions(+), 18 deletions(-) rename flowfile_core/flowfile_core/alembic/versions/{012_secret_access_events.py => 015_secret_access_events.py} (95%) create mode 100644 flowfile_core/tests/test_master_key_normalization.py create mode 100644 shared/crypto/master_key.py diff --git a/docs/for-developers/secret-storage.md b/docs/for-developers/secret-storage.md index a1b439a6e..5d078a1e0 100644 --- a/docs/for-developers/secret-storage.md +++ b/docs/for-developers/secret-storage.md @@ -99,7 +99,7 @@ Table: `secret_access_events`. | `result_status` | string | `success` / `error` | | `error` | text, nullable | short snake_case codes: `not_found`, `duplicate_name`, … | | `source` | string, default `api` | reserved for future non-API emitters | -| `ip_address` | string, nullable | best-effort; honors `X-Forwarded-For` | +| `ip_address` | string, nullable | best-effort; direct connection IP by default. `X-Forwarded-For` is honored only when `FLOWFILE_TRUST_PROXY_HEADERS` is set — trustworthy only behind a proxy that overwrites the header. | | `created_at` | datetime, indexed | server-side `now()` | ### Recording an event from a new route diff --git a/docs/users/visual-editor/catalog/secrets.md b/docs/users/visual-editor/catalog/secrets.md index 736c70c38..bd8e960c9 100644 --- a/docs/users/visual-editor/catalog/secrets.md +++ b/docs/users/visual-editor/catalog/secrets.md @@ -22,10 +22,13 @@ Three short ideas, in order. **Master key.** One Fernet key per Flowfile instance. Every encrypted secret in your database is unlockable by this key — and only this key. If it's lost, the encrypted secrets are unrecoverable; there is no backdoor. -**Per-user derivation.** Each user's secrets are encrypted with a key *derived* from the master key plus that user's ID (HKDF-SHA256). User A's derived key cannot decrypt User B's secrets, even though both come from the same master key. This isolates accidental cross-user reads at the cryptographic level. +**Per-user derivation.** Each user's secrets are encrypted with a key *derived* from the master key plus that user's ID (HKDF-SHA256). This is **defense-in-depth, not an access boundary**: anyone holding the master key can derive every user's key and decrypt any user's secret — the envelope even embeds the `user_id`. The boundaries that actually keep users apart are (a) keeping the master key secret and (b) the `user_id` filter applied to every secret query at the database layer. Per-user derivation only buys you something if a single *derived* key leaks without the master key. **Envelope format.** Each encrypted value in the database is stored as `$ffsec$1$$`. The `1` is a version digit reserved for future format changes. The `` lets the worker process decrypt without being separately told who the secret belongs to. +!!! warning "The worker holds the master key too" + Both `flowfile_core` and `flowfile_worker` read the same master key, so the worker can decrypt **any** user's secret. Compromise of the worker is therefore equivalent to full secret compromise. Harden it the same way you harden core: no plaintext-secret logging, network isolation (the worker should not be exposed beyond core), and least-privilege for its host and credentials. + ## Where the master key lives | Mode | Location | @@ -107,9 +110,12 @@ Every secret create / list / read / delete attempt — successful or failed — - The action (`create`, `list`, `read`, `delete`) - The secret name (when applicable) - Result status (`success` or `error`) and error code (e.g. `not_found`, `duplicate_name`) -- Source IP address (best-effort, honors `X-Forwarded-For`) +- Source IP address (best-effort; see note below) - Timestamp +!!! note "When is the recorded IP trustworthy?" + By default Flowfile records the direct connection IP and **ignores** `X-Forwarded-For`, because a directly-exposed instance lets any client forge that header. Set `FLOWFILE_TRUST_PROXY_HEADERS=true` only when Flowfile sits behind a reverse proxy that *overwrites* `X-Forwarded-For` — then the recorded `ip_address` reflects the real client. Without such a proxy, treat `ip_address` as advisory, not forensic. + Admins can query the log via `GET /secrets/secrets/audit` (with optional `secret_name`, `action`, and `limit` query parameters). Non-admins see only their own events. **What is *not* recorded:** per-secret decrypt events during flow execution. Logging every decrypt would dominate the table with routine activity. If you need decrypt-time accounting later, that's a separate feature. diff --git a/flowfile_core/flowfile_core/alembic/versions/012_secret_access_events.py b/flowfile_core/flowfile_core/alembic/versions/015_secret_access_events.py similarity index 95% rename from flowfile_core/flowfile_core/alembic/versions/012_secret_access_events.py rename to flowfile_core/flowfile_core/alembic/versions/015_secret_access_events.py index 29215bb52..ba04e3dbd 100644 --- a/flowfile_core/flowfile_core/alembic/versions/012_secret_access_events.py +++ b/flowfile_core/flowfile_core/alembic/versions/015_secret_access_events.py @@ -11,8 +11,8 @@ investigate the deletion). ``secret_name`` is denormalized into the audit row for the same reason. -Revision ID: 012 -Revises: 011 +Revision ID: 015 +Revises: 014 Create Date: 2026-05-12 """ @@ -21,8 +21,8 @@ import sqlalchemy as sa from alembic import op -revision: str = "012" -down_revision: str | None = "011" +revision: str = "015" +down_revision: str | None = "014" branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None diff --git a/flowfile_core/flowfile_core/auth/secrets.py b/flowfile_core/flowfile_core/auth/secrets.py index 5739effef..a34388893 100644 --- a/flowfile_core/flowfile_core/auth/secrets.py +++ b/flowfile_core/flowfile_core/auth/secrets.py @@ -10,6 +10,8 @@ from cryptography.fernet import Fernet +from shared.crypto.master_key import normalize_master_key + logger = logging.getLogger(__name__) @@ -160,7 +162,7 @@ def get_docker_secret_key() -> str | None: """ env_key = os.environ.get("FLOWFILE_MASTER_KEY") if env_key: - env_key = env_key.strip().strip('"').strip("'") + env_key = normalize_master_key(env_key) try: Fernet(env_key.encode()) return env_key @@ -171,7 +173,7 @@ def get_docker_secret_key() -> str | None: if os.path.isfile(secret_path): try: with open(secret_path) as f: - key = f.read().strip() + key = normalize_master_key(f.read()) Fernet(key.encode()) return key except Exception: diff --git a/flowfile_core/flowfile_core/routes/secrets.py b/flowfile_core/flowfile_core/routes/secrets.py index 707c7502d..6c508458e 100644 --- a/flowfile_core/flowfile_core/routes/secrets.py +++ b/flowfile_core/flowfile_core/routes/secrets.py @@ -32,13 +32,16 @@ def _client_ip(request: Request) -> str | None: """Best-effort client IP extraction. - Prefers the forwarded-for header when the deployment sits behind a proxy, - falls back to the direct connection. Returned as a string or ``None`` if - nothing is available (e.g. in some test clients). + ``X-Forwarded-For`` is honored only when ``FLOWFILE_TRUST_PROXY_HEADERS`` is + set (the deployment sits behind a proxy that overwrites the header); + otherwise the header is ignored and the direct connection IP is used, since + a directly-exposed instance would let clients forge the recorded source IP. + Returns ``None`` if nothing is available (e.g. in some test clients). """ - forwarded = request.headers.get("x-forwarded-for") - if forwarded: - return forwarded.split(",")[0].strip() + if os.environ.get("FLOWFILE_TRUST_PROXY_HEADERS", "").strip().lower() in ("1", "true", "yes", "on"): + forwarded = request.headers.get("x-forwarded-for") + if forwarded: + return forwarded.split(",")[0].strip() client = request.client return client.host if client else None diff --git a/flowfile_core/flowfile_core/secret_manager/audit.py b/flowfile_core/flowfile_core/secret_manager/audit.py index 7b52b9b97..721a044b4 100644 --- a/flowfile_core/flowfile_core/secret_manager/audit.py +++ b/flowfile_core/flowfile_core/secret_manager/audit.py @@ -69,8 +69,8 @@ def record_event(event: SecretEvent, db: Session | None = None) -> SecretAccessE try: if db is not None: - db.add(row) - db.flush() + with db.begin_nested(): + db.add(row) else: with SessionLocal() as session: session.add(row) diff --git a/flowfile_core/tests/test_master_key_normalization.py b/flowfile_core/tests/test_master_key_normalization.py new file mode 100644 index 000000000..f9052645b --- /dev/null +++ b/flowfile_core/tests/test_master_key_normalization.py @@ -0,0 +1,28 @@ +"""F2: core and worker must resolve a quoted/whitespace FLOWFILE_MASTER_KEY to the same key. + +``/setup/generate-key`` instructs operators to write ``FLOWFILE_MASTER_KEY=""`` +into ``.env``. If the surrounding quotes survive into the env value, core and worker +must agree on the resulting key or the worker can't decrypt what core encrypted. +""" + +import os +import sys + +from cryptography.fernet import Fernet + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..")) + +from flowfile_core.auth.secrets import get_docker_secret_key as core_get_docker_secret_key +from flowfile_worker.secrets import get_docker_secret_key as worker_get_docker_secret_key + + +def test_quoted_master_key_resolves_identically_in_core_and_worker(monkeypatch): + valid_key = Fernet.generate_key().decode() + monkeypatch.setenv("FLOWFILE_MASTER_KEY", f'"{valid_key}"') + + core_key = core_get_docker_secret_key() + worker_key = worker_get_docker_secret_key() + + assert core_key == valid_key + assert worker_key == valid_key + assert core_key == worker_key diff --git a/flowfile_core/tests/test_secret_audit.py b/flowfile_core/tests/test_secret_audit.py index e2048c8ae..c2803336d 100644 --- a/flowfile_core/tests/test_secret_audit.py +++ b/flowfile_core/tests/test_secret_audit.py @@ -138,6 +138,25 @@ def test_audit_endpoint_returns_recent_events(): client.delete(f"/secrets/secrets/{name}") +def test_list_endpoint_survives_audit_write_failure(monkeypatch): + """F3: a failing audit insert must not poison the session and 500 the read.""" + from sqlalchemy.orm import Session + + client = _get_client() + + real_flush = Session.flush + + def failing_flush(self, *args, **kwargs): + if any(isinstance(obj, SecretAccessEvent) for obj in self.new): + raise RuntimeError("simulated audit flush failure") + return real_flush(self, *args, **kwargs) + + monkeypatch.setattr(Session, "flush", failing_flush) + + r = client.get("/secrets/secrets") + assert r.status_code == 200, r.text + + def test_audit_endpoint_filters_by_action(): client = _get_client() name = f"audit_filter_{uuid.uuid4().hex[:8]}" diff --git a/flowfile_worker/flowfile_worker/secrets.py b/flowfile_worker/flowfile_worker/secrets.py index e8b3b42ff..acc2cdf9f 100644 --- a/flowfile_worker/flowfile_worker/secrets.py +++ b/flowfile_worker/flowfile_worker/secrets.py @@ -26,6 +26,7 @@ from shared.crypto.envelope import ( derive_user_key as _derive_user_key, ) +from shared.crypto.master_key import normalize_master_key logger = logging.getLogger(__name__) @@ -113,6 +114,7 @@ def get_docker_secret_key() -> str | None: """ env_key = os.environ.get("FLOWFILE_MASTER_KEY") if env_key: + env_key = normalize_master_key(env_key) try: Fernet(env_key.encode()) return env_key @@ -124,7 +126,7 @@ def get_docker_secret_key() -> str | None: if os.path.exists(secret_path): try: with open(secret_path) as f: - key = f.read().strip() + key = normalize_master_key(f.read()) Fernet(key.encode()) return key except Exception as e: diff --git a/flowfile_worker/tests/test_secrets.py b/flowfile_worker/tests/test_secrets.py index 327eba7fa..83491dcb0 100644 --- a/flowfile_worker/tests/test_secrets.py +++ b/flowfile_worker/tests/test_secrets.py @@ -78,6 +78,18 @@ def test_invalid_env_key_raises(self, monkeypatch): with pytest.raises(RuntimeError, match="not a valid Fernet key"): get_docker_secret_key() + def test_quoted_env_key_is_normalized(self, monkeypatch): + from cryptography.fernet import Fernet + valid_key = Fernet.generate_key().decode() + monkeypatch.setenv("FLOWFILE_MASTER_KEY", f'"{valid_key}"') + assert get_docker_secret_key() == valid_key + + def test_whitespace_env_key_is_normalized(self, monkeypatch): + from cryptography.fernet import Fernet + valid_key = Fernet.generate_key().decode() + monkeypatch.setenv("FLOWFILE_MASTER_KEY", f" {valid_key}\n") + assert get_docker_secret_key() == valid_key + class TestDeriveUserKey: """Test derive_user_key function.""" diff --git a/shared/crypto/__init__.py b/shared/crypto/__init__.py index 2088f4882..7856e8bf4 100644 --- a/shared/crypto/__init__.py +++ b/shared/crypto/__init__.py @@ -15,6 +15,7 @@ encrypt_secret_envelope, parse_v1_envelope, ) +from shared.crypto.master_key import normalize_master_key __all__ = [ "KEY_DERIVATION_VERSION", @@ -22,5 +23,6 @@ "decrypt_secret_envelope", "derive_user_key", "encrypt_secret_envelope", + "normalize_master_key", "parse_v1_envelope", ] diff --git a/shared/crypto/master_key.py b/shared/crypto/master_key.py new file mode 100644 index 000000000..6e69a7b7b --- /dev/null +++ b/shared/crypto/master_key.py @@ -0,0 +1,18 @@ +"""Master-key string normalization shared by core and worker. + +A ``FLOWFILE_MASTER_KEY`` value commonly arrives wrapped in quotes or padded +with whitespace, because ``/setup/generate-key`` tells operators to write +``FLOWFILE_MASTER_KEY=""`` into ``.env`` and some shells/loaders keep the +surrounding quotes. Core and worker must agree byte-for-byte on how that raw +value maps to a Fernet key, so the normalization lives here and both services +call it on every key source (env var and Docker secret file). +""" + + +def normalize_master_key(raw: str) -> str: + """Strip surrounding whitespace and matching single/double quotes. + + A Fernet key is URL-safe base64, so it never legitimately contains quote + characters — stripping them from both ends can't corrupt a valid key. + """ + return raw.strip().strip('"').strip("'")