diff --git a/Cargo.lock b/Cargo.lock index 616afc56..fdef188f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1926,6 +1926,7 @@ dependencies = [ "sqlx", "tokio", "tracing-subscriber", + "url", "uuid", ] diff --git a/Cargo.toml b/Cargo.toml index afbb5048..90b7dda7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,6 +60,10 @@ chrono = "0.4" # stays on native-tls with no rustls/ring/aws-lc in the tree. reqwest = { version = "0.13", default-features = false, features = ["json", "native-tls", "multipart"] } +# For redacting credential-bearing URLs before they reach a log or an error. +# Already in the graph via reqwest; declared directly so redact.rs can use it. +url = "2.5" + # For df.http_multipart() — base64-decodes part payloads carried in the config JSON. base64 = "0.22" diff --git a/USER_GUIDE.md b/USER_GUIDE.md index 053be94f..612e3759 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -1042,6 +1042,10 @@ Durable function variables allow you to configure durable functions with externa > **Important**: `df.setvar()`, `df.unsetvar()`, and `df.clearvars()` cannot be called from within a running durable function. They are for configuration only. +> **Variables are not secrets.** `df.vars` stores values as plaintext, and the value is copied +> into durable execution history when `df.start()` runs. See +> [Variables and secrets](#variables-and-secrets) before putting a credential in one. + ### System Variables These read-only variables are automatically available during durable function execution: @@ -1051,6 +1055,39 @@ These read-only variables are automatically available during durable function ex | `{sys_instance_id}` | Current durable function instance ID | | `{sys_label}` | Durable function label (if provided) | +### Variables and secrets + +Using `df.setvar()` to hold an API key is a natural move, and it genuinely helps — but only in +one place. Here is exactly what it does and does not protect, so you can make an informed +choice. + +Given this pattern: + +```sql +SELECT df.setvar('api_key', ''); +SELECT df.start(df.http('https://api.example.com/users', 'GET', NULL, + '{"Authorization": "Bearer {api_key}"}'::jsonb), 'fetch-users'); +``` + +| Location | Holds the credential? | Notes | +|----------|----------------------|-------| +| `df.nodes.query` | No — stores the literal `{api_key}` | This is the one thing `{var}` genuinely buys you. Substitution happens at execution time, not when the node is created. | +| `df.vars.value` | **Yes, plaintext** | RLS isolates it from other users. It is *not* encrypted, and it is not hidden from a superuser or the table owner. | +| Durable execution history | **Yes, plaintext** | `df.start()` snapshots every variable you own into the orchestration input, and the substituted header is recorded as the HTTP step's input. Parallel branches and each loop iteration re-record it. | +| WAL, backups, replicas | **Yes** | Follows every write above, typically with a longer retention than `pg_durable.retention_days`. | +| Server log | URLs are redacted; SQL text depends on `pg_durable.log_workflow_sql` | See [What reaches the server log](#what-reaches-the-server-log). | +| `pg_stat_activity`, `pg_stat_statements` | Only if you inline the literal | `SELECT df.setvar('api_key', '')` is itself a statement. Bind it as a parameter, or use `\getenv` in psql, rather than typing the value into SQL. | + +Writing the credential directly into `df.http(...)` instead of using a variable is strictly +worse: it adds `df.nodes.query` to that list without removing anything from it. + +Practical guidance: + +- Prefer credentials that are short-lived and narrowly scoped, so history exposure is bounded. +- Treat `df.vars` as configuration — hostnames, table names, batch sizes, API versions. +- Treat a database backup as containing every credential any workflow has used. +- Set `pg_durable.retention_days` to the shortest value your operations allow. + ### Variable Substitution > **Security note**: All `{...}` substitutions — including `{varname}`, `{sys_label}`, and `{sys_instance_id}` — perform **raw text substitution**. The value is inserted directly into the SQL string without escaping or parameterization. This is by design so that variables can hold SQL fragments like table names or expressions. Since you control both the variable value and the query template, and SQL executes under your own role, this is safe for configuration values you set yourself. Do **not** store untrusted external input in variables that get substituted into SQL. For passing query *results* between steps, use `$name` (via `|=>`), which applies proper SQL escaping. @@ -1060,11 +1097,11 @@ Use `{varname}` in SQL queries to substitute variable values: ```sql -- Set up configuration SELECT df.setvar('api_base', 'https://api.example.com'); -SELECT df.setvar('api_key', 'secret123'); +SELECT df.setvar('page_size', '50'); -- Start durable function using variables SELECT df.start( - df.http('{api_base}/users', 'GET', NULL, '{"Authorization": "Bearer {api_key}"}'::jsonb) + df.http('{api_base}/users?per_page={page_size}', 'GET') ~> 'INSERT INTO playground.logs (msg) VALUES (''Fetched users'')', 'fetch-users' ); @@ -2060,10 +2097,38 @@ Row-level security (RLS) restricts each user to their own instances and nodes: - `df.cancel()` and `df.signal()` check ownership before acting — attempts on other users' instances return "Instance not found or access denied" - Superusers bypass RLS and can see all instances (standard PostgreSQL behavior) +### What reaches the server log + +The background worker writes an execution trace to the PostgreSQL server log. That log is a +different security boundary from the database: RLS does not apply to it, it is not bound by +`pg_durable.retention_days`, and it is often shipped and backed up separately. + +| Trace | Contents | +|-------|----------| +| HTTP and multipart requests | Method, scheme, host, port and path. **Query-string values, userinfo and the URL fragment are redacted**, so an Azure SAS token or `?api-key=` does not reach the log. Parameter *names* are kept so the line stays diagnosable. | +| HTTP request errors | Same redaction, applied to the message text as well, since the HTTP client interpolates the request URL into its own errors. | +| HTTP and multipart request headers and bodies | Never logged. | +| SQL nodes | The submitting role and target database, plus the fully-substituted SQL text when `pg_durable.log_workflow_sql` is on. | +| Workflow result | The final return value is logged on completion. For a workflow ending in an HTTP step this includes the response body. | + +`pg_durable.log_workflow_sql` (default `on`) is what gates the SQL text. SQL cannot be +redacted heuristically — a variable value spliced into a query is indistinguishable from the +query itself — so this is an on/off switch rather than a masking rule. It is read by the +background worker, so like the other worker GUCs it is Postmaster-context and needs a restart: + +```ini +# postgresql.conf — omit workflow SQL text from the log +pg_durable.log_workflow_sql = off +``` + +Turning it off also removes the primary record of what workflows actually executed, which is the +first thing an incident investigation looks for. Leave it on unless the server log is less well +protected than the database, and prefer keeping credentials out of SQL in the first place. + ### Security Best Practices 1. **Worker role must be superuser** — The background worker role (`pg_durable.worker_role`) must be a superuser to bypass RLS and manage all instances -2. **Review df.vars usage** — Variables are scoped per-user via RLS, but avoid storing secrets in plain text +2. **Do not put credentials in `df.vars`** — Variables are scoped per-user via RLS, but they are stored as plaintext and are copied into durable execution history. See [Variables and secrets](#variables-and-secrets) 3. **Use labels carefully** — Instance labels are visible only to the submitting user (RLS-filtered) and superusers 4. **Monitor instances** — Superusers can use `df.list_instances()` to see all users' instances; regular users see only their own 5. **Avoid unsafe `SECURITY DEFINER` wrappers around `df.start()`** — Never allow untrusted callers to supply SQL or futures to `df.start()` from a `SECURITY DEFINER` context unless definer-level execution is intentional. @@ -2166,7 +2231,7 @@ GRANT pg_durable_user TO app_backend, etl_service; Users get `SELECT` and `INSERT` on `df.instances` and `df.nodes` (required for `df.start()`, `df.status()`, `df.result()`). Column-level `UPDATE` on `(status, updated_at)` allows `df.cancel()` to set status. No full `UPDATE` or `DELETE` — the identity column (`submitted_by`) and structural columns are protected. -> **Note:** `df.vars` uses per-user scoping via an `owner` column and RLS — each user can only read and write their own variables. Superusers bypass RLS but the DSL functions (`df.setvar()`, `df.getvar()`, etc.) still scope to the calling user via explicit filters. Avoid storing secrets in plain text. +> **Note:** `df.vars` uses per-user scoping via an `owner` column and RLS — each user can only read and write their own variables. Superusers bypass RLS but the DSL functions (`df.setvar()`, `df.getvar()`, etc.) still scope to the calling user via explicit filters. Values are stored as plaintext and are copied into durable execution history — see [Variables and secrets](#variables-and-secrets). ### Revoking Privileges @@ -2247,7 +2312,7 @@ pg_durable.max_new_transaction_starts = 2 pg_durable.new_transaction_start_timeout = 5 ``` -> **Other GUCs:** `pg_durable.list_instances_max_limit` (SUSET context, default `1000`) caps the per-call page size of `df.list_instances()`. Unlike the connection-limit GUCs above, it is superuser-settable at runtime (no restart) and is not loaded from `postgresql.conf` at startup only. See [docs/api-reference.md](docs/api-reference.md#pg_durablelist_instances_max_limit). +> **Other GUCs:** `pg_durable.list_instances_max_limit` (SUSET context, default `1000`) caps the per-call page size of `df.list_instances()`. Unlike the connection-limit GUCs above, it is superuser-settable at runtime (no restart) and is not loaded from `postgresql.conf` at startup only. See [docs/api-reference.md](docs/api-reference.md#pg_durablelist_instances_max_limit). `pg_durable.log_workflow_sql` (Postmaster context, default `on`) controls whether workflow SQL text is written to the server log — see [What reaches the server log](#what-reaches-the-server-log). ### Connection Budget Formula diff --git a/docs/api-reference.md b/docs/api-reference.md index e74aad1e..41e9510c 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -588,6 +588,11 @@ Sets a workflow variable for the current user (before `df.start()`). Each user h SELECT df.setvar('api_url', 'https://api.example.com'); ``` +> **Not for credentials.** Values are stored as plaintext in `df.vars`, and `df.start()` copies +> every variable you own into durable execution history. Using `{varname}` keeps the value out of +> `df.nodes.query` but not out of history. See +> [Variables and secrets](../USER_GUIDE.md#variables-and-secrets). + --- ### df.getvar(name) @@ -684,7 +689,7 @@ SELECT df.revoke_usage('app_role'); ## Server Configuration (GUCs) -These settings are configured via `ALTER SYSTEM SET` or `postgresql.conf` and take effect after `SELECT pg_reload_conf()` (no restart required). +These settings are configured via `ALTER SYSTEM SET` or `postgresql.conf`. Each one lists its context: `SUSET` settings take effect after `SELECT pg_reload_conf()`, while `POSTMASTER` settings require a restart. Every GUC read by the background worker is `POSTMASTER`, because the worker does not process a configuration reload. --- @@ -749,3 +754,28 @@ SELECT pg_reload_conf(); ``` > **Behavior change (v0.2.4):** prior to v0.2.4, `df.list_instances()` silently truncated `limit_count` to 10000. It now raises an error when `limit_count` exceeds this GUC (default 1000). Callers that previously requested very large pages should lower `limit_count` or use the paginated overload (`after_cursor`/`next_cursor`). + +--- + +### pg_durable.log_workflow_sql + +Controls whether the background worker writes the SQL text of an executed workflow node to the PostgreSQL server log. + +| Property | Value | +|----------|-------| +| Type | `boolean` | +| Default | `on` | +| Context | `POSTMASTER` (set in `postgresql.conf`; requires restart) | + +The SQL is logged *after* variable substitution, so a credential held in a `df.vars` variable and spliced into a query reaches the server log in cleartext. Unlike a URL query string — which pg_durable redacts unconditionally — SQL cannot be masked heuristically, because a substituted value is indistinguishable from the surrounding statement. This GUC is therefore an on/off switch. + +```ini +# postgresql.conf +pg_durable.log_workflow_sql = off +``` + +The value is read by the background worker, which does not process a configuration reload, so a restart is required — the same as `pg_durable.retention_days` and the connection-limit GUCs. + +Turning it off keeps the submitting role and target database in the log but drops the statement text. That also removes the primary record of what workflows executed, which is usually the first thing an incident investigation looks for — prefer keeping credentials out of SQL over disabling the log. See [Variables and secrets](../USER_GUIDE.md#variables-and-secrets). + +This trace is written by the background worker's own logging, not by PostgreSQL statement logging, so `log_statement` neither enables nor suppresses it. diff --git a/docs/http-security.md b/docs/http-security.md index 860d3f8b..f623a5af 100644 --- a/docs/http-security.md +++ b/docs/http-security.md @@ -166,22 +166,25 @@ REVOKE EXECUTE ON FUNCTION df.http(text, text, text, jsonb, integer) FROM PUBLIC ``` When `df.grant_usage(role, include_http => false)` is called and the role still -has effective HTTP access via the PUBLIC grant (or another inherited grant), a -`WARNING` is emitted to signal that the revocation had no net effect. +has effective HTTP access via the PUBLIC grant (or another inherited grant), +that grant is left intact — `df.grant_usage()` is purely additive and never +issues a `REVOKE`. Call `df.revoke_usage()` first to downgrade a role. ### 3.4 Admin function protection `df.grant_usage()` and `df.revoke_usage()` are admin-only functions. `EXECUTE` is revoked from `PUBLIC` at `CREATE EXTENSION` time, so only -superusers can call them. - -> **Caution:** `df.grant_usage()` internally runs -> `GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA df`, which temporarily includes -> `df.grant_usage()` and `df.revoke_usage()` themselves before the function -> immediately revokes them from the target role. If an admin replicates the -> blanket `GRANT` manually without the matching `REVOKE`s, the target role -> will gain access to these admin helpers. Always use `df.grant_usage()` -> rather than hand-crafting the equivalent `GRANT` statements. +superusers can call them — plus any role an admin delegated to with +`df.grant_usage(role, with_grant => true)`. + +`df.grant_usage()` issues a specific set of grants: `USAGE ON SCHEMA df`, +column-scoped table privileges, `EXECUTE` on `df.http()` / `df.http_multipart()` +when `include_http => true`, and `EXECUTE` on `df.grant_usage()`, +`df.revoke_usage()` and `df.metrics()` when `with_grant => true`. Those five +functions are the only ones with `PUBLIC` `EXECUTE` revoked at install; every +other `df.*` function keeps PostgreSQL's default, so **schema `USAGE` is the +real access gate**. Granting `USAGE ON SCHEMA df` by hand therefore exposes the +whole DSL surface at once; prefer `df.grant_usage()`. ### 3.5 Feature-flag interaction @@ -315,12 +318,47 @@ Every HTTP attempt (allowed or blocked) is logged via `ctx.trace_info` with: - `submitted_by` — the role that called `df.start()` at the time the node was created (captured as `current_user` in the DSL and stored in `FunctionNode`) -- `url` — the requested URL +- `url` — the requested URL, **redacted** (see below) - Block reason tag — `(scheme)`, `(allowlist)`, or `(ip)` in the log prefix Resolved IP addresses are **not** included in error messages or logs to avoid leaking internal network topology to potentially malicious users. +### 7.1 URL redaction + +A URL is a credential carrier: an Azure SAS token lives entirely in the query +string, and `?api-key=` / `?code=` are common elsewhere. The server log is a +weaker boundary than the database — no RLS, no `pg_durable.retention_days`, and +frequently a separate shipping and backup path — so `crate::redact::redact_url` +is applied before any URL reaches a log line or an error string. + +| Component | Treatment | +|-----------|-----------| +| Scheme, host, port, path | Preserved. The URL is reparsed, so a logged line may be normalized (host lowercased, default port dropped) relative to what the workflow supplied. | +| Query parameter *names* | Preserved — `sig` and `code` are not themselves secret, and keeping them makes a redacted line diagnosable | +| Query parameter *values* | Replaced with ``, except an allowlist of benign parameters (`api-version`, `apiversion`, `comp`, `restype`) | +| Bare query token with no `=` | Replaced whole — indistinguishable from a name | +| `userinfo@` | Replaced with `@`, keeping the host | +| Fragment | Replaced whole | +| Unparseable input | Replaced whole — redaction fails closed and never echoes back a string it could not parse | + +Parsing uses the `url` crate, so IPv6 authorities, percent-encoding, default +ports and userinfo follow the spec rather than ad-hoc string splitting. + +The same redaction is applied to the HTTP client's own error text, which +interpolates the request URL into messages such as +`error sending request for url (...)`. This matters because activity errors are +persisted to `df.nodes.error` and recorded in durable execution history, not +just written to the log. + +Request headers and bodies are never logged. + +> **Not covered:** the response body. A workflow's final result is traced on +> completion, and the full response envelope is stored in `df.nodes.result`. A +> request whose *response* is a credential — reading a secret from Key Vault, or +> an OAuth token endpoint — still persists that value. Response-side redaction +> is tracked separately. + --- ## 8. Error Messages diff --git a/scripts/test-e2e-local.sh b/scripts/test-e2e-local.sh index aacdb60c..c43b30e2 100755 --- a/scripts/test-e2e-local.sh +++ b/scripts/test-e2e-local.sh @@ -268,21 +268,41 @@ DATA_DIR="$PGRX_HOME/data-$PG_VERSION" LOG_FILE="$PGRX_HOME/$PG_VERSION.log" CONF_FILE="$DATA_DIR/postgresql.conf" -shopt -s nullglob -PGRX_CANDIDATES=("$PGRX_HOME"/"$PG_VERSION".*/pgrx-install/bin) -shopt -u nullglob -if [ "${#PGRX_CANDIDATES[@]}" -eq 0 ]; then +# config.toml is authoritative: it is what cargo pgrx builds the extension +# against, and it may point at an externally installed PostgreSQL (pgenv, a +# distro package) that has no directory under ~/.pgrx. Fall back to the +# pgrx-built layout when the entry is missing. +PGRX_BIN="" +if [ -f "$PGRX_HOME/config.toml" ]; then + PG_CONFIG_PATH=$(grep -E "^pg${PG_VERSION}[[:space:]]*=[[:space:]]*\"" "$PGRX_HOME/config.toml" | head -1 | cut -d'"' -f2) + if [ -n "$PG_CONFIG_PATH" ]; then + PGRX_BIN="$(dirname "$PG_CONFIG_PATH")" + fi +fi + +if [ -z "$PGRX_BIN" ]; then + shopt -s nullglob + PGRX_CANDIDATES=("$PGRX_HOME"/"$PG_VERSION".*/pgrx-install/bin) + shopt -u nullglob + if [ "${#PGRX_CANDIDATES[@]}" -gt 0 ]; then + PGRX_BIN="${PGRX_CANDIDATES[0]}" + fi +fi + +if [ -z "$PGRX_BIN" ] || [ ! -x "$PGRX_BIN/pg_ctl" ]; then echo "Error: pgrx PostgreSQL $PG_VERSION not installed" echo "Run: cargo pgrx init" exit 1 fi -PGRX_BIN="${PGRX_CANDIDATES[0]}" PSQL="$PGRX_BIN/psql" PG_CTL="$PGRX_BIN/pg_ctl" PG_ISREADY="$PGRX_BIN/pg_isready" PG_CONFIG="$PGRX_BIN/pg_config" +# Keep a developer ~/.psqlrc out of test output, so local runs match CI. +export PSQLRC=/dev/null + stop_server() { if [ -d "$DATA_DIR" ] && "$PG_CTL" status -D "$DATA_DIR" >/dev/null 2>&1; then echo -e "${YELLOW}Stopping PostgreSQL...${NC}" diff --git a/scripts/test-upgrade.sh b/scripts/test-upgrade.sh index 6975e314..4dea5646 100755 --- a/scripts/test-upgrade.sh +++ b/scripts/test-upgrade.sh @@ -62,9 +62,23 @@ done PGRX_HOME="$HOME/.pgrx" PG_PORT="$((28800 + PG_VERSION))" -# Find pgrx binaries -PGRX_BIN=$(ls -d "$PGRX_HOME/$PG_VERSION."*/pgrx-install/bin 2>/dev/null | head -1) +# Find pgrx binaries. config.toml is authoritative: it is what cargo pgrx builds +# the extension against, and it may point at an externally installed PostgreSQL +# (pgenv, a distro package) with no directory under ~/.pgrx. Fall back to the +# pgrx-built layout when the entry is missing. +PGRX_BIN="" +if [ -f "$PGRX_HOME/config.toml" ]; then + PG_CONFIG_PATH=$(grep -E "^pg${PG_VERSION}[[:space:]]*=[[:space:]]*\"" "$PGRX_HOME/config.toml" | head -1 | cut -d'"' -f2) + if [ -n "$PG_CONFIG_PATH" ]; then + PGRX_BIN="$(dirname "$PG_CONFIG_PATH")" + fi +fi + if [ -z "$PGRX_BIN" ]; then + PGRX_BIN=$(ls -d "$PGRX_HOME/$PG_VERSION."*/pgrx-install/bin 2>/dev/null | head -1) +fi + +if [ -z "$PGRX_BIN" ] || [ ! -x "$PGRX_BIN/pg_ctl" ]; then echo "Error: pgrx PostgreSQL $PG_VERSION not installed" echo "Run: cargo pgrx init" exit 1 @@ -76,6 +90,10 @@ PG_ISREADY="$PGRX_BIN/pg_isready" PG_CONFIG="$PGRX_BIN/pg_config" DATA_DIR="$PGRX_HOME/data-$PG_VERSION" LOG_FILE="$PGRX_HOME/$PG_VERSION.log" + +# A developer ~/.psqlrc echoes each \pset it runs, which corrupts the exact-match +# assertions below. Point psql at an empty rc file instead. +export PSQLRC=/dev/null EXTENSION_DIR=$("$PG_CONFIG" --sharedir)/extension # Version detection: read current version from Cargo.toml diff --git a/src/activities/execute_http.rs b/src/activities/execute_http.rs index a98feb8c..007b0e96 100644 --- a/src/activities/execute_http.rs +++ b/src/activities/execute_http.rs @@ -96,6 +96,12 @@ pub async fn execute( .as_deref() .ok_or("Blocked: HTTP node has no submitted_by \u{2014} cannot verify privilege")?; + // Every log line and error below reports this, never `config.url`: an Azure + // SAS token lives entirely in the query string, and both sinks outlive the + // instance (the server log has no retention bound; errors are persisted to + // df.nodes.error and into duroxide history). + let safe_url = crate::redact::redact_url(&config.url); + // Validation chain — order is security-critical: // 0. Privilege: submitted_by must hold EXECUTE on df.http(). Closes the // bypass path where a user crafts raw Durofut JSON and passes @@ -114,31 +120,28 @@ pub async fn execute( .await .inspect_err(|_| { ctx.trace_info(format!( - "HTTP BLOCKED (privilege) url={} submitted_by={audit_user}", - config.url + "HTTP BLOCKED (privilege) url={safe_url} submitted_by={audit_user}" )); })?; // --- Scheme validation (always enforced, regardless of feature flag) --- crate::ssrf::validate_url_scheme(&config.url).inspect_err(|_| { ctx.trace_info(format!( - "HTTP BLOCKED (scheme) url={} submitted_by={audit_user}", - config.url + "HTTP BLOCKED (scheme) url={safe_url} submitted_by={audit_user}" )); })?; // --- Azure endpoint allow-list (blocks all bare IPs + non-Azure domains) --- crate::ssrf::validate_url_allowlist(&config.url).inspect_err(|_| { ctx.trace_info(format!( - "HTTP BLOCKED (allowlist) url={} submitted_by={audit_user}", - config.url + "HTTP BLOCKED (allowlist) url={safe_url} submitted_by={audit_user}" )); })?; let start = std::time::Instant::now(); ctx.trace_info(format!( - "HTTP {} {} submitted_by={audit_user}", - config.method, config.url + "HTTP {} {safe_url} submitted_by={audit_user}", + config.method )); // Build client with SSRF-safe resolver (when feature enabled) and timeout @@ -178,10 +181,9 @@ pub async fn execute( // a structured audit log (mirrors the scheme-block log above). if crate::ssrf::is_ssrf_block_error(&err_string) { ctx.trace_info(format!( - "HTTP BLOCKED (ip) url={} submitted_by={audit_user}", - config.url + "HTTP BLOCKED (ip) url={safe_url} submitted_by={audit_user}" )); - return err_string; + return crate::redact::redact_urls_in(&err_string); } // Try to extract status code from error if available @@ -190,21 +192,22 @@ pub async fn execute( .map(|s| format!(" (HTTP {})", s.as_u16())) .unwrap_or_default(); + // reqwest's Display interpolates the request URL, so the error text is + // scrubbed as well as the URL we format ourselves. + let detail = crate::redact::redact_urls_in(&err_string); + if e.is_timeout() { format!( "HTTP timeout after {}s{}: {}", - config.timeout_seconds, status_info, config.url + config.timeout_seconds, status_info, safe_url ) } else if e.is_connect() { - format!( - "HTTP connection failed{}: {} - {}", - status_info, config.url, e - ) + format!("HTTP connection failed{status_info}: {safe_url} - {detail}") } else if e.is_status() { // Error due to HTTP status code - format!("HTTP request failed{}: {} - {}", status_info, config.url, e) + format!("HTTP request failed{status_info}: {safe_url} - {detail}") } else { - format!("HTTP request failed{}: {} - {}", status_info, config.url, e) + format!("HTTP request failed{status_info}: {safe_url} - {detail}") } })?; @@ -237,9 +240,8 @@ pub async fn execute( // Fail on 5xx server errors (transient, should retry) if status.is_server_error() { return Err(format!( - "HTTP {} {} returned {}: {}", + "HTTP {} {safe_url} returned {}: {}", config.method, - config.url, status_code, response_body.error_preview() )); diff --git a/src/activities/execute_multipart.rs b/src/activities/execute_multipart.rs index 422b3c9a..153417f5 100644 --- a/src/activities/execute_multipart.rs +++ b/src/activities/execute_multipart.rs @@ -91,6 +91,9 @@ pub async fn execute( "Blocked: HTTP_MULTIPART node has no submitted_by \u{2014} cannot verify privilege", )?; + // See execute_http: never log or return `config.url` itself. + let safe_url = crate::redact::redact_url(&config.url); + // Validation chain — order is security-critical and mirrors execute_http: // 0. Privilege: submitted_by must hold EXECUTE on df.http_multipart(). // 1. Scheme: blocks file://, gopher://, etc. @@ -103,32 +106,28 @@ pub async fn execute( .await .inspect_err(|_| { ctx.trace_info(format!( - "HTTP_MULTIPART BLOCKED (privilege) url={} submitted_by={audit_user}", - config.url + "HTTP_MULTIPART BLOCKED (privilege) url={safe_url} submitted_by={audit_user}" )); })?; // --- Scheme validation (always enforced) --- crate::ssrf::validate_url_scheme(&config.url).inspect_err(|_| { ctx.trace_info(format!( - "HTTP_MULTIPART BLOCKED (scheme) url={} submitted_by={audit_user}", - config.url + "HTTP_MULTIPART BLOCKED (scheme) url={safe_url} submitted_by={audit_user}" )); })?; // --- Azure endpoint allow-list --- crate::ssrf::validate_url_allowlist(&config.url).inspect_err(|_| { ctx.trace_info(format!( - "HTTP_MULTIPART BLOCKED (allowlist) url={} submitted_by={audit_user}", - config.url + "HTTP_MULTIPART BLOCKED (allowlist) url={safe_url} submitted_by={audit_user}" )); })?; let start = std::time::Instant::now(); ctx.trace_info(format!( - "HTTP_MULTIPART {} {} ({} parts) submitted_by={audit_user}", + "HTTP_MULTIPART {} {safe_url} ({} parts) submitted_by={audit_user}", config.method, - config.url, config.parts.len() )); @@ -191,10 +190,9 @@ pub async fn execute( // Detect SSRF IP-blocklist rejections from the resolver. if crate::ssrf::is_ssrf_block_error(&err_string) { ctx.trace_info(format!( - "HTTP_MULTIPART BLOCKED (ip) url={} submitted_by={audit_user}", - config.url + "HTTP_MULTIPART BLOCKED (ip) url={safe_url} submitted_by={audit_user}" )); - return err_string; + return crate::redact::redact_urls_in(&err_string); } let status_info = e @@ -202,18 +200,18 @@ pub async fn execute( .map(|s| format!(" (HTTP {})", s.as_u16())) .unwrap_or_default(); + // reqwest's Display interpolates the request URL — scrub it too. + let detail = crate::redact::redact_urls_in(&err_string); + if e.is_timeout() { format!( "HTTP timeout after {}s{}: {}", - config.timeout_seconds, status_info, config.url + config.timeout_seconds, status_info, safe_url ) } else if e.is_connect() { - format!( - "HTTP connection failed{}: {} - {}", - status_info, config.url, e - ) + format!("HTTP connection failed{status_info}: {safe_url} - {detail}") } else { - format!("HTTP request failed{}: {} - {}", status_info, config.url, e) + format!("HTTP request failed{status_info}: {safe_url} - {detail}") } })?; @@ -246,9 +244,8 @@ pub async fn execute( // Fail on 5xx server errors (transient, should retry) if status.is_server_error() { return Err(format!( - "HTTP_MULTIPART {} {} returned {}: {}", + "HTTP_MULTIPART {} {safe_url} returned {}: {}", config.method, - config.url, status_code, response_body.error_preview() )); diff --git a/src/activities/execute_sql.rs b/src/activities/execute_sql.rs index 29e7edeb..62b11c9e 100644 --- a/src/activities/execute_sql.rs +++ b/src/activities/execute_sql.rs @@ -193,14 +193,18 @@ pub async fn execute( serde_json::from_str(&input_json).map_err(|e| format!("Invalid execute_sql input: {e}"))?; ctx.trace_info(format!( - "Executing SQL as '{}'{}: {}", + "Executing SQL as '{}'{}{}", input.submitted_by, input .database .as_ref() .map(|db| format!(" in database '{db}'")) .unwrap_or_default(), - input.query + if crate::types::log_workflow_sql_enabled() { + format!(": {}", input.query) + } else { + String::new() + } )); // Acquire a permit from the user-connection semaphore. The permit is held diff --git a/src/lib.rs b/src/lib.rs index 11736117..6a79ec34 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -51,6 +51,15 @@ pub static RETENTION_DAYS: GucSetting = GucSetting::::new(30); /// terminal instances and reclaims orphaned engine records. `0` disables it. pub static RECONCILE_INTERVAL: GucSetting = GucSetting::::new(3600); +/// When `false`, the worker log omits the SQL text of executed workflow nodes. +/// The text is logged fully substituted, so a `{var}` holding a credential is +/// written to the server log in cleartext. Unlike a query string, SQL cannot be +/// redacted heuristically, so this is on/off rather than a masking rule. +/// +/// Postmaster context: it is read in the background worker, which never calls +/// `ProcessConfigFile`, so a reload would not reach it. +pub static LOG_WORKFLOW_SQL: GucSetting = GucSetting::::new(true); + // Module declarations pub mod activities; pub mod client; @@ -59,6 +68,7 @@ pub mod explain; pub mod monitoring; pub mod node_status; pub mod orchestrations; +pub mod redact; pub mod registry; pub mod ssrf; pub mod types; @@ -219,6 +229,15 @@ pub extern "C-unwind" fn _PG_init() { GucFlags::default(), ); + GucRegistry::define_bool_guc( + c"pg_durable.log_workflow_sql", + c"Log the SQL text of executed workflow nodes to the worker log", + c"The SQL is logged after variable substitution, so any credential held in a df.vars variable and spliced into a query is written to the PostgreSQL server log in cleartext. Set to off in environments where the server log is less protected than the database. Turning this off also removes the primary forensic record of what workflows executed. Requires server restart to change.", + &LOG_WORKFLOW_SQL, + GucContext::Postmaster, + GucFlags::default(), + ); + worker::register_background_worker(); } diff --git a/src/orchestrations/execute_function_graph.rs b/src/orchestrations/execute_function_graph.rs index d2e42dbb..491efc07 100644 --- a/src/orchestrations/execute_function_graph.rs +++ b/src/orchestrations/execute_function_graph.rs @@ -269,6 +269,8 @@ pub async fn execute(ctx: OrchestrationContext, input_json: String) -> Result { + // FIXME: This might be tracing sensitive data. Should it be fixed? + // It's a larger behavioral change than the rest of this work. ctx.trace_info(format!("Function completed with result: {result}")); let status_input = serde_json::json!({ "instance_id": input.instance_id, @@ -596,7 +598,10 @@ async fn execute_sql_node( .ok_or_else(|| format!("SQL node {node_id} has no query"))?; let final_query = substitute_all(query, results, &exec_ctx.vars, sys_vars)?; - ctx.trace_info(format!("Executing SQL: {final_query}")); + // The substituted text is not traced here: the execute_sql activity logs it + // under pg_durable.log_workflow_sql, and an orchestration must not read a + // GUC to decide what to emit. + ctx.trace_info(format!("Executing SQL node {node_id}")); let input = serde_json::json!({ "query": final_query, @@ -1567,7 +1572,12 @@ async fn execute_http_node( let final_config = config.to_string(); let url = config["url"].as_str().unwrap_or("?"); let method = config["method"].as_str().unwrap_or("POST"); - ctx.trace_info(format!("Executing HTTP {method} {url}")); + // Substitution has already run, so `url` may carry a SAS token or api-key. + // redact_url is pure, so tracing it stays replay-deterministic. + ctx.trace_info(format!( + "Executing HTTP {method} {}", + crate::redact::redact_url(url) + )); let result = ctx .schedule_activity(activities::execute_http::NAME, final_config) @@ -1666,7 +1676,10 @@ async fn execute_http_multipart_node( let final_config = config.to_string(); let url = config["url"].as_str().unwrap_or("?"); let method = config["method"].as_str().unwrap_or("POST"); - ctx.trace_info(format!("Executing HTTP_MULTIPART {method} {url}")); + ctx.trace_info(format!( + "Executing HTTP_MULTIPART {method} {}", + crate::redact::redact_url(url) + )); let result = ctx .schedule_activity(activities::execute_multipart::NAME, final_config) diff --git a/src/redact.rs b/src/redact.rs new file mode 100644 index 00000000..0f40b52a --- /dev/null +++ b/src/redact.rs @@ -0,0 +1,333 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the PostgreSQL License. + +//! Redaction of credential-bearing text before it reaches a log or an error. +//! +//! A URL is a credential carrier: Azure SAS tokens live entirely in the query +//! string, and `?api-key=`/`?code=` are common elsewhere. The worker's tracing +//! subscriber writes at `info` by default (see `worker::init_tracing`), so an +//! unredacted URL lands in the PostgreSQL server log in cleartext — a sink with +//! no RLS, no `pg_durable.retention_days`, and often a different backup path +//! than the database itself. Error strings are worse still: they are persisted +//! to `df.nodes.error` and into duroxide history. +//! +//! Redaction is deliberately lossy and fails closed: anything that cannot be +//! parsed as a URL is replaced wholesale rather than echoed back. + +use url::Url; + +/// Stand-in for any elided value. Deliberately not a fixed-width mask, so the +/// length of the original is not disclosed. +pub const REDACTED: &str = ""; + +/// Query parameters whose values carry no credential and are worth keeping for +/// diagnosis. Matched case-insensitively against the parameter name. +const SAFE_QUERY_PARAMS: &[&str] = &["api-version", "apiversion", "comp", "restype"]; + +/// Characters that cannot appear inside a URL, used to find where an embedded +/// URL ends when scanning free-form text. +/// +/// `{` and `}` are deliberately absent so that a URL still carrying a `{var}` +/// placeholder is scanned as a single token rather than being cut in half. +const URL_TERMINATORS: &[char] = &['"', '\'', '<', '>', '\\', '^', '`', '|', '(', ')', ',']; + +fn is_safe_query_param(name: &str) -> bool { + SAFE_QUERY_PARAMS + .iter() + .any(|safe| name.eq_ignore_ascii_case(safe)) +} + +/// Redact the values in a raw query string, preserving parameter names. +/// +/// This splits on `&` and `=` rather than using [`Url::query_pairs`], because +/// `query_pairs` follows the form-urlencoded rules and reports a bare token +/// (`?SEKRIT`, no `=`) as the *name* of a valueless parameter. Emitting that +/// would publish the token verbatim. Here a bare token is elided whole, since a +/// lone token is indistinguishable from a name. +fn redact_query(query: &str) -> String { + let mut out = String::with_capacity(query.len()); + + for (i, pair) in query.split('&').enumerate() { + if i > 0 { + out.push('&'); + } + match pair.split_once('=') { + // An empty value has nothing to elide, and leaving it alone is what + // makes redaction idempotent: re-redacting `sig=` must not + // append a second marker. + Some((_, "")) => out.push_str(pair), + Some((name, _)) if is_safe_query_param(name) => out.push_str(pair), + Some((name, _)) => { + out.push_str(name); + out.push('='); + out.push_str(REDACTED); + } + None => out.push_str(REDACTED), + } + } + + out +} + +/// Redact the credential-bearing parts of a single URL. +/// +/// Preserved: scheme, host, port, path. Elided: userinfo, every query-parameter +/// value except [`SAFE_QUERY_PARAMS`], and the fragment. +/// +/// The output is rebuilt from the parsed components rather than by mutating the +/// [`Url`], because the setters percent-encode `<` and `>` and would turn the +/// marker into `%3Credacted%3E`. +pub fn redact_url(url: &str) -> String { + let Ok(parsed) = Url::parse(url) else { + // Not parseable — never echo it back. + return REDACTED.to_string(); + }; + + let mut out = String::with_capacity(url.len()); + out.push_str(parsed.scheme()); + out.push_str("://"); + + // Keep only the fact that credentials were present, not which. + if !parsed.username().is_empty() || parsed.password().is_some() { + out.push_str(REDACTED); + out.push('@'); + } + + if let Some(host) = parsed.host_str() { + out.push_str(host); + } + if let Some(port) = parsed.port() { + out.push(':'); + out.push_str(&port.to_string()); + } + + out.push_str(parsed.path()); + + if let Some(query) = parsed.query().filter(|q| !q.is_empty()) { + out.push('?'); + out.push_str(&redact_query(query)); + } + + if parsed.fragment().is_some() { + out.push('#'); + out.push_str(REDACTED); + } + + out +} + +/// Redact every URL embedded in free-form text. +/// +/// Needed because `reqwest::Error`'s `Display` interpolates the request URL, so +/// wrapping a transport error without scrubbing it would reintroduce the very +/// query string [`redact_url`] was applied to remove. +/// +/// Locating a URL inside prose is necessarily heuristic — a parser can validate +/// a candidate but cannot tell you where one ends in surrounding text. So this +/// only delimits candidates; [`redact_url`] does the parsing, and a candidate it +/// rejects is replaced wholesale. +pub fn redact_urls_in(text: &str) -> String { + if !text.contains("://") { + return text.to_string(); + } + + let mut out = String::with_capacity(text.len()); + let mut rest = text; + + while let Some(sep) = rest.find("://") { + // Walk back over the scheme, which must be alphanumeric with `+-.`. + let scheme_start = rest[..sep] + .rfind(|c: char| !(c.is_ascii_alphanumeric() || c == '+' || c == '-' || c == '.')) + .map_or(0, |i| i + 1); + + if scheme_start == sep { + // `://` with no scheme in front of it — not a URL. + out.push_str(&rest[..sep + 3]); + rest = &rest[sep + 3..]; + continue; + } + + out.push_str(&rest[..scheme_start]); + + let candidate = &rest[scheme_start..]; + let end = candidate + .find(|c: char| c.is_whitespace() || URL_TERMINATORS.contains(&c)) + .unwrap_or(candidate.len()); + + out.push_str(&redact_url(&candidate[..end])); + rest = &candidate[end..]; + } + + out.push_str(rest); + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn redacts_sas_token_query_string() { + let redacted = redact_url( + "https://acct.blob.core.windows.net/c/b?sv=2022-11-02&ss=b&sig=abc%2Fdef%3D", + ); + assert_eq!( + redacted, + "https://acct.blob.core.windows.net/c/b?sv=&ss=&sig=" + ); + assert!(!redacted.contains("abc")); + } + + #[test] + fn keeps_safe_query_params() { + assert_eq!( + redact_url("https://v.vault.azure.net/secrets/s?api-version=7.4&code=SEKRIT"), + "https://v.vault.azure.net/secrets/s?api-version=7.4&code=" + ); + } + + #[test] + fn safe_query_param_match_is_case_insensitive() { + assert_eq!( + redact_url("https://h/p?API-Version=7.4"), + "https://h/p?API-Version=7.4" + ); + } + + #[test] + fn preserves_scheme_host_port_and_path() { + assert_eq!( + redact_url("https://host.example.com:8443/a/b/c"), + "https://host.example.com:8443/a/b/c" + ); + } + + #[test] + fn redacts_userinfo_but_keeps_host() { + assert_eq!( + redact_url("https://user:pa%40ss@host.example.com/p"), + "https://@host.example.com/p" + ); + } + + #[test] + fn redacts_fragment() { + assert_eq!( + redact_url("https://h/p#access_token=SEKRIT"), + "https://h/p#" + ); + assert_eq!( + redact_url("https://h/p?a=1#access_token=SEKRIT"), + "https://h/p?a=#" + ); + } + + #[test] + fn redacts_bare_query_token_whole() { + assert_eq!(redact_url("https://h/p?SEKRIT"), "https://h/p?"); + } + + #[test] + fn handles_bracketed_ipv6_authority() { + assert_eq!( + redact_url("http://[2001:db8::1]:8080/p?k=v"), + "http://[2001:db8::1]:8080/p?k=" + ); + } + + #[test] + fn handles_authority_only_urls() { + // An empty path normalizes to "/" per the WHATWG URL spec. + assert_eq!(redact_url("https://host"), "https://host/"); + assert_eq!(redact_url("https://host?k=v"), "https://host/?k="); + assert_eq!(redact_url("https://host#f"), "https://host/#"); + assert_eq!(redact_url("https://host/"), "https://host/"); + } + + #[test] + fn empty_query_does_not_emit_question_mark() { + assert_eq!(redact_url("https://h/p?"), "https://h/p"); + } + + #[test] + fn empty_parameter_value_is_left_alone() { + assert_eq!( + redact_url("https://h/p?a=&b=SEKRIT"), + "https://h/p?a=&b=" + ); + } + + #[test] + fn fails_closed_on_non_url_input() { + assert_eq!(redact_url(""), REDACTED); + assert_eq!(redact_url("not a url"), REDACTED); + assert_eq!(redact_url("/relative/path?sig=SEKRIT"), REDACTED); + } + + #[test] + fn keeps_unsubstituted_placeholders_legible() { + // A URL whose {var} placeholders were never substituted still reaches + // redaction. Nothing is lost: the host keeps its placeholder verbatim + // and the path is percent-encoded, so the mistake is still diagnosable. + assert_eq!( + redact_url("https://{kv_host}/secrets/{name}?api-version=7.4"), + "https://{kv_host}/secrets/%7Bname%7D?api-version=7.4" + ); + } + + #[test] + fn bare_query_token_is_not_treated_as_a_parameter_name() { + // Url::query_pairs follows the form-urlencoded rules and reports a bare + // token as the NAME of a valueless parameter. Rebuilding the query from + // those pairs would publish the token verbatim, which is why + // redact_query splits on '&'/'=' itself. Pin the hazard so a future + // refactor to query_pairs fails loudly here. + let parsed = Url::parse("https://h/p?SEKRIT").unwrap(); + let names: Vec<_> = parsed.query_pairs().map(|(k, _)| k.into_owned()).collect(); + assert_eq!(names, vec!["SEKRIT".to_string()]); + + assert_eq!(redact_url("https://h/p?SEKRIT"), "https://h/p?"); + } + + #[test] + fn redacts_url_embedded_in_error_text() { + let scrubbed = redact_urls_in( + "error sending request for url (https://h/p?sig=SEKRIT): connection closed", + ); + assert_eq!( + scrubbed, + "error sending request for url (https://h/p?sig=): connection closed" + ); + assert!(!scrubbed.contains("SEKRIT")); + } + + #[test] + fn redacts_every_url_in_text() { + let scrubbed = redact_urls_in("from https://a/x?k=S1 to https://b/y?k=S2 failed"); + assert!(!scrubbed.contains("S1")); + assert!(!scrubbed.contains("S2")); + assert_eq!( + scrubbed, + "from https://a/x?k= to https://b/y?k= failed" + ); + } + + #[test] + fn leaves_text_without_urls_untouched() { + assert_eq!(redact_urls_in("plain error, no url"), "plain error, no url"); + assert_eq!(redact_urls_in(""), ""); + } + + #[test] + fn tolerates_bare_scheme_separator() { + assert_eq!(redact_urls_in("weird :// text"), "weird :// text"); + } + + #[test] + fn redaction_is_idempotent() { + let once = redact_url("https://h/p?sig=SEKRIT"); + assert_eq!(redact_url(&once), once); + let once_in_text = redact_urls_in("url (https://h/p?sig=SEKRIT)"); + assert_eq!(redact_urls_in(&once_in_text), once_in_text); + } +} diff --git a/src/types.rs b/src/types.rs index 00ab16c3..3c9eee58 100644 --- a/src/types.rs +++ b/src/types.rs @@ -88,6 +88,11 @@ pub fn get_reconcile_interval() -> Duration { Duration::from_secs(crate::RECONCILE_INTERVAL.get() as u64) } +/// Returns `true` when the SQL text of a workflow node may be written to the log. +pub fn log_workflow_sql_enabled() -> bool { + crate::LOG_WORKFLOW_SQL.get() +} + /// Returns `true` when superuser-submitted instances are permitted. pub fn superuser_instances_enabled() -> bool { crate::ENABLE_SUPERUSER_INSTANCES.get()