diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 00789612..5cff5a4a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -219,6 +219,8 @@ jobs: sudo apt-get update sudo apt-get install -y \ pkg-config \ + openssl \ + python3 \ libssl-dev \ libclang-dev \ clang \ diff --git a/CHANGELOG.md b/CHANGELOG.md index 3221202b..068a3648 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,21 @@ All notable changes to this project are documented in this file. The format is b Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may include breaking changes. +## [Unreleased] + +### Added + +- **Managed identity HTTP authentication:** normal and multipart endpoints can + use a single user-assigned identity configured at startup with + `pg_durable.managed_identity_client_id`. An empty setting disables MI without + affecting other HTTP authentication. Tokens are acquired and cached inside + activities, with audiences derived from supported destination hosts. +- **Delegated MI administration:** `EXECUTE` on `df.managed_identity_admin()` + permits endpoint configuration without PostgreSQL superuser. The endpoint + owner's capability and the caller's endpoint/HTTP privileges are checked for + each attempt. Upgrade each installation using MI to 0.2.10 and grant the + capability explicitly; existing general administration grants do not include it. + ## [0.2.9] - 2026-10-07 > **Upgrade:** Install/restart the new binary and wait for the control runtime diff --git a/Cargo.lock b/Cargo.lock index 998df174..13daf3b3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1915,7 +1915,7 @@ dependencies = [ [[package]] name = "pg_durable" -version = "0.2.9" +version = "0.2.10" dependencies = [ "base64 0.23.1", "bigdecimal", diff --git a/Cargo.toml b/Cargo.toml index ba9cd10a..c6a7eee5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "pg_durable" -version = "0.2.9" +version = "0.2.10" edition = "2021" license = "PostgreSQL" repository = "https://github.com/microsoft/pg_durable" diff --git a/USER_GUIDE.md b/USER_GUIDE.md index ff65cabd..78733a2f 100644 --- a/USER_GUIDE.md +++ b/USER_GUIDE.md @@ -969,8 +969,9 @@ no round trip through a table — see [Multipart Uploads](#multipart-uploads). ### Endpoint Credential Catalog Endpoint definitions use a handler-less `pg_durable_fdw`. A foreign server holds -the base URL and authentication scheme; each caller's user mapping holds its -credentials. There are no foreign tables or scans. +the base URL and authentication scheme. Static credentials belong in each caller's +user mapping; managed identity obtains tokens at execution time without a mapping. +There are no foreign tables or scans. `auth_scheme` is required. `base_url` may be omitted only with `auth_scheme 'none'` for a server used solely for named secrets. A supplied base URL must be a nonempty, @@ -984,10 +985,11 @@ a server does not authorize network access or bypass HTTP destination restrictio | `bearer` | None | `token` (without the `Bearer ` prefix) | | `header` | `header_name`, such as `x-api-key` | `header_value` | | `query` | None | `query_string`, already URL-encoded, optionally starting with `?` | +| `managed-identity` | None; uses the startup-configured identity | None | -Unknown options and authentication schemes are rejected. `managed-identity` is -reserved and rejected until its authentication controls are available. The -catalog does not accept free-form `resource`, `scope` or `client_id` settings. +Unknown options and authentication schemes are rejected. The catalog does not +accept `client_id`, `resource` or `scope` settings. Identity selection belongs to +deployment configuration, not individual endpoints. Header names cannot override routing, framing or multipart content type. Credential values must be nonempty and valid for their transport; validation errors do not echo those values. The mapping validator checks individual options; @@ -1024,8 +1026,9 @@ Server owners can change the destination, so they must be trusted with credentia sent through their servers. Catalog masking does not prevent an owner from redirecting a subsequent request to a destination they control. -The catalog resolver checks server `USAGE` and reads the authenticated caller's -mapping on every attempt. Rotation affects the next lookup, not an already-sent +The catalog resolver checks server `USAGE` on every attempt and reads the +authenticated caller's mapping when the scheme requires it. Rotation affects the +next lookup, not an already-sent request. Missing or inaccessible credentials fail without privileged fallback. Mappings also accept individual `secret.` options for explicit bindings. These are separate from `token`, `header_value` and `query_string`; other option @@ -1039,6 +1042,112 @@ in provisioning DDL can appear in PostgreSQL logs. Treat backup/restore and credential provisioning accordingly. Dropping the extension with `CASCADE` also removes dependent servers and mappings. +### Managed Identity + +**Added in 0.2.10.** Install and restart the new binary, then run +`ALTER EXTENSION pg_durable UPDATE TO '0.2.10'` in each database where MI +endpoints will be configured. The 0.2.9 endpoint API does not include MI's +administration capability. + +Managed identity authenticates both normal and multipart endpoint requests with +an Azure access token acquired inside the HTTP activity. Tokens are not exposed +through `df.secret`, stored in user mappings, or added to workflow inputs or +results. A destination can still return sensitive data, including an echoed +request header; response contents are not automatically redacted. + +The server administrator selects one user-assigned identity for all MI-backed +endpoints, then restarts PostgreSQL: + +```ini +pg_durable.managed_identity_client_id = '11111111-1111-1111-1111-111111111111' +``` + +The value must be a nonzero client UUID. An empty value, the default, disables MI +requests without affecting other HTTP authentication. There is no implicit +system-assigned or provider-default identity. The identity must be available to +the token provider and authorized for the target resources. Select an identity +intended for customer-configured requests, not merely one used by a service for +internal operations. + +Creating or altering an MI endpoint requires `EXECUTE` on +`df.managed_identity_admin()`, in addition to native FDW and ownership permissions. +This is a database-local capability with no `PUBLIC` grant. Calling the function +returns no token and changes no settings. `df.grant_usage`, even with +`with_grant => true`, does not grant this capability or FDW creation permission. +Endpoint definitions and capability grants belong to the workflow's origin +installation. A grant in the control database does not grant MI administration +in satellites. The startup identity and provider settings apply to the shared +worker and therefore to all installations it serves. + +A privileged provisioner can delegate configuration to an existing administrator +role without making it a PostgreSQL superuser: + +```sql +GRANT USAGE ON SCHEMA df TO endpoint_admin; +GRANT USAGE ON FOREIGN DATA WRAPPER pg_durable_fdw TO endpoint_admin; +GRANT EXECUTE ON FUNCTION df.managed_identity_admin() TO endpoint_admin; +SELECT df.grant_usage('app_role', include_http => true); + +SET ROLE endpoint_admin; +CREATE SERVER storage_identity FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://account.blob.core.windows.net', + auth_scheme 'managed-identity'); +GRANT USAGE ON FOREIGN SERVER storage_identity TO app_role; +RESET ROLE; + +SET ROLE app_role; +SELECT df.start( + df.http(df.endpoint('storage_identity', '/container/blob'), 'GET', + headers => '{"x-ms-version":"2023-11-03"}'), + 'read-blob' +); +RESET ROLE; +``` + +Each request rechecks that the endpoint owner retains the capability and that the +caller has server `USAGE` and HTTP execution privileges. Revoking the owner's +capability, or transferring ownership to an unauthorized role, blocks subsequent +attempts even if a token is cached. Native privilege inheritance applies; another +direct or inherited grant can still supply the capability. Revocation does not +recall an already-authorized request. + +Endpoint administration does not permit changing the client-ID or provider GUCs. +Endpoint options, workflow paths, headers, secret bindings and user mappings cannot +select another identity, change its token resource, or replace its `Authorization` +header. Pinning the identity controls pg_durable's requests, not host-wide access +to credentials; provider isolation remains a deployment responsibility. + +The destination must use HTTPS on port 443. Token resources are fixed by hostname: + +| Destination | Token resource | +|---|---| +| `*.openai.azure.com`, `*.cognitiveservices.azure.com`, `*.services.ai.azure.com` | `https://cognitiveservices.azure.com` | +| `*.blob.core.windows.net`, `*.blob.storage.azure.net`, `*.dfs.core.windows.net`, `*.queue.core.windows.net`, `*.table.core.windows.net`, `*.file.core.windows.net` | `https://storage.azure.com/` | +| `*.vault.azure.net` | `https://vault.azure.net` | +| `management.azure.com` | `https://management.azure.com/` | + +Other hosts, custom application audiences and sovereign-cloud endpoints are not +supported for managed identity. The HTTP security mode, domain allow-list and DNS/IP +checks still apply. Some mapped hosts, including `management.azure.com`, require +an explicit addition to `pg_durable.http_allowed_domains`. + +By default, tokens come from the public [Azure IMDS token endpoint](https://learn.microsoft.com/en-us/entra/identity/managed-identities-azure-resources/how-to-use-vm-token). +No Azure connection is made at startup or for other authentication schemes. +The same binary works outside Azure; selecting managed identity there requires an +available token provider. There is no automatic CLI, environment-credential or +alternate-provider fallback. If a deployment does not expose IMDS, its administrator +must supply an IMDS-compatible adapter and configure the startup-only +[`pg_durable.managed_identity_endpoint`](docs/api-reference.md#pg_durablemanaged_identity_endpoint). +Such adapters are not included in the extension. + +The worker caches one token per resource for the configured identity, refreshes before expiry, and +never returns a stale token after a refresh failure. Permissions and destination +configuration are still checked on every request, including cache hits. Provider +failures fail the activity without persisting the provider's response body. +Changing the client ID and restarting clears the cache and changes the identity +used by subsequent requests, including pending workflows and retries. Recorded +activity results replay without acquiring tokens. + ### Calling an Endpoint Pass `df.endpoint(server, path)` as the destination of either HTTP constructor: diff --git a/docs/E2E_TESTING.md b/docs/E2E_TESTING.md index 0f5c3e2d..b2200337 100644 --- a/docs/E2E_TESTING.md +++ b/docs/E2E_TESTING.md @@ -73,6 +73,7 @@ The test suite is organized into 23 files. Files `01`–`09` open with `SET SESS | `16_heartbeat.sql` | Worker heartbeat liveness — `df._worker_epoch.last_seen_at` advances over time | | `52_node_id_collision_across_instances.sql` | Cross-instance node-ID collision — two instances own the same 8-hex node id; asserts composite-PK coexistence, that `(instance_id, id)` addresses exactly one row, `df.result()` is instance-scoped, and a scoped `update_node_status`-style UPDATE affects exactly one row (issue #129) | | `68_long_caller_transaction.sql` | Caller-transaction handoff beyond five seconds, transient graph-probe failure, whole rollback, and savepoint rollback | +| `76_managed_identity_unconfigured.sql` | Unset identity disables MI for both HTTP forms while ordinary HTTP still succeeds | ### Startup-Phase Specific @@ -89,6 +90,22 @@ restart with the phase's configuration, not a rebuild. | `48_http_allow_all.sql` | `http-allow-all` | Unrestricted mode admits plaintext HTTP, unlisted domains, and private destinations for both HTTP activities | | `69_http_allowed_domains.sql` | `http-custom-domains` | Custom allow-list enforcement, malformed startup rejection, protected mode settings, and restart-only policy changes | | `70_http_allowed_domains_empty.sql` | `http-empty-domains` | Empty restricted-mode allow-list denies all domains | +| `75_managed_identity.sql` | `managed-identity` | Offline startup-pinned UAMI requests, delegated ownership, revocation with a warm cache, override rejection, restart-only configuration and token non-persistence | + +The managed-identity phase needs Bash 4+, Python 3 and the `openssl` command, but +no Azure account or network service. The runner starts the local token and TLS +fixtures in [managed_identity_mock.py](../tests/e2e/managed_identity_mock.py), +generates a temporary certificate, and removes the fixtures on exit. It selects +`pg_durable.http_security = 'unrestricted'` to route the approved hostname through a local HTTPS +proxy; production code has no mock-provider bypass. The custom-domain and +HTTP-disabled phases separately verify that managed identity cannot override +those policies. With `--keep`, PostgreSQL remains available for state inspection, +but the identity fixtures stop when the runner exits. The separate unconfigured +test runs in the standard phase and uses the usual `httpbingo.org` HTTP probe. + +```bash +./scripts/test-e2e-local.sh managed_identity +``` ## Test Structure diff --git a/docs/api-reference.md b/docs/api-reference.md index b4c477d6..1ce268d5 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -356,7 +356,8 @@ the FDW validator invoked by PostgreSQL on creation and alteration; it returns `void` or raises an error without echoing credential values. The server options are `base_url`, `auth_scheme`, and `header_name` (only for -header authentication). `auth_scheme` is required; `base_url` may be omitted only +header authentication). `client_id`, `resource` and `scope` are not accepted. +`auth_scheme` is required; `base_url` may be omitted only with `auth_scheme 'none'` for named-secret storage. A supplied URL retains all validation requirements, and endpoint requests fail explicitly if it is absent. Mapping options are `token`, `header_value`, @@ -365,6 +366,43 @@ native per-option `ADD`, `SET` and `DROP`. See [Endpoint Credential Catalog](../USER_GUIDE.md#endpoint-credential-catalog) for option combinations, grants, rotation and backup implications. +Since v0.2.10, `auth_scheme 'managed-identity'` requires `EXECUTE` on +`df.managed_identity_admin()` for configuration and for the endpoint owner at +execution. It requires no user mapping for authentication and uses the single +identity pinned by `pg_durable.managed_identity_client_id` at startup. +The destination must be a supported public Azure hostname over HTTPS on port 443; +the token resource is derived from that host, never supplied by the workflow. +Server `USAGE`, HTTP function permissions and destination policy remain mandatory. +See [Managed Identity](../USER_GUIDE.md#managed-identity) for supported hosts and +deployment requirements. + +### df.managed_identity_admin() + +**Added in 0.2.10.** Update the extension in the database where the endpoint is +defined before granting this capability. + +An extension-owned, no-op function returning `void`. Its `EXECUTE` privilege +authorizes managed-identity endpoint configuration; calling it never acquires a +token or changes configuration. There is no `PUBLIC` execution grant. + +```sql +GRANT EXECUTE ON FUNCTION df.managed_identity_admin() TO endpoint_admin; +REVOKE EXECUTE ON FUNCTION df.managed_identity_admin() FROM endpoint_admin; +``` + +Native FDW and ownership privileges remain required. The capability does not +authorize changing provider settings, imply HTTP execution privileges, or permit +selecting a different identity. It is not included in `df.grant_usage` or +`df.revoke_usage`; manage it explicitly with `GRANT` and `REVOKE`. + +The endpoint owner's effective `EXECUTE` privilege is rechecked on every attempt, +including token-cache hits. A direct or inherited grant counts; `NOINHERIT` +memberships alone do not. Losing the capability or transferring ownership to an +unauthorized role blocks subsequent attempts, not already-authorized requests. +A missing or non-extension-owned capability function fails closed. +The capability and endpoints are local to the workflow's origin installation; +control-database grants do not authorize satellite endpoint administration. + ### df.http_multipart(url [, method, parts, headers, timeout]) Makes an HTTP request with a `multipart/form-data` body. Requires the same @@ -1013,6 +1051,74 @@ database, not the explicit SQL target. Excess callers wait up to --- +### pg_durable.managed_identity_client_id + +The user-assigned identity used by all managed-identity endpoint requests. +Available since v0.2.10. The shared worker uses this setting for control and +satellite requests alike; it is not a per-database identity setting. + +| Property | Value | +|----------|-------| +| Type | `string`, containing a nonzero client UUID or empty | +| Default | Empty (managed identity disabled) | +| Context | `POSTMASTER` (requires a PostgreSQL restart) | +| Visibility | Superusers and roles with `pg_read_all_settings` | + +```ini +pg_durable.managed_identity_client_id = '11111111-1111-1111-1111-111111111111' +``` + +An unset or empty value makes MI requests fail before contacting a token provider. +Other HTTP authentication is unaffected. Whitespace, malformed UUIDs and the nil +UUID are rejected, including at server startup. Automatic system-assigned or +provider-default identity selection is not supported. + +Configure this in `postgresql.conf` or through an authorized `ALTER SYSTEM SET`. +Session, role and database overrides are rejected; reload alone does not apply a +change. MI endpoint administration grants do not confer configuration privileges. +Changing the value and restarting changes the identity for subsequent attempts, +including pending workflows and retries, and clears the token cache. The configured +identity is not added to workflow inputs; recorded results replay unchanged. + +--- + +### pg_durable.managed_identity_endpoint + +The token-provider URL for endpoint managed-identity authentication. Available +since v0.2.10; it is unused by other authentication schemes. + +| Property | Value | +|----------|-------| +| Type | `string` | +| Default | `http://169.254.169.254/metadata/identity/oauth2/token` | +| Context | `POSTMASTER` (requires a PostgreSQL restart) | +| Visibility | Superusers and roles with configuration-reading privileges | + +The provider must implement the public IMDS GET token protocol: `Metadata: true`, +`api-version=2018-02-01`, a host-derived `resource`, and the configured `client_id`. +The provider must honor that identity selection rather than silently substituting +another identity. +Success must include `access_token`, `token_type` (`Bearer`), the requested +`resource`, and an `expires_on` Unix timestamp or `expires_in` seconds. Expiry +values may be JSON numbers or strings. Tokens must have more than two minutes +remaining and must already be valid. + +Only administrators configure this URL. It must be HTTPS, or HTTP with a literal +loopback or IMDS IP address, without userinfo, query or fragment. Session, role and +database overrides are rejected. An invalid startup setting prevents startup. +An adapter for a different identity service must be provided separately; there is +no provider auto-discovery or fallback on failure. + +Token requests use a separate client with no redirects or environment proxies, +a two-second connection timeout, a ten-second total acquisition deadline +(including cache waits), and a 64 KiB response limit. Failures expose only local +diagnostics or HTTP status, not the provider's body. The worker holds one cache +entry per supported token resource for its pinned identity and coalesces +concurrent fetches for each entry. +The cache is in memory only and is cleared by worker restart. + +--- + ### pg_durable.enable_superuser_instances Controls whether pg_durable allows durable function instances whose `submitted_by` role is a PostgreSQL superuser. diff --git a/docs/http-security.md b/docs/http-security.md index c5e0b779..6dbf2320 100644 --- a/docs/http-security.md +++ b/docs/http-security.md @@ -274,7 +274,7 @@ destination policy. HTTP constructors are not decoded as endpoint references. An endpoint value does not grant authority. Normal and multipart activities first check their existing HTTP function grant, then resolve the foreign server -and the submitting role's user mapping on a connection authenticated as that role. +and, when needed, the submitting role's user mapping on a connection authenticated as that role. Server `USAGE` is mandatory. Catalogs are in the trusted origin installation (control for legacy control-origin work), regardless of the SQL target. Caller-supplied database/identity fields in node JSON cannot select another credential catalog or @@ -351,6 +351,66 @@ examples and deferred general-composition cases. --- +### 3.8 Managed identity + +Managed identity is available with the 0.2.10 binary and extension schema. +Capability and endpoint checks use the trusted origin installation, including +satellite-local grants. The identity and token-provider startup settings apply +to the shared worker, not separately to each database. + +Managed-identity endpoints delegate a deployment-selected identity, not a per-role +catalog credential. Creation and alteration require `EXECUTE` on the +extension-owned `df.managed_identity_admin()` permission function, alongside +native FDW and ownership privileges. The function is a no-op, not a token API, +and its ACL has no `PUBLIC` grant. General pg_durable administration does not +implicitly grant this capability. + +Activities independently check that the current endpoint owner retains effective +`EXECUTE` on that same extension-owned function. A missing or detached function +fails closed; a same-named user function does not establish authority. Native +privilege inheritance applies. Losing the capability or transferring ownership +to an unauthorized role disables subsequent use, including cache hits. Server +`USAGE` and HTTP function grants are also checked before consulting the token +cache. Revocation does not retract a request already authorized and in flight. + +Only explicitly mapped public Azure hosts over HTTPS on port 443 are accepted. +The resource is fixed by the destination host; neither server options nor workflow +data can supply a free-form resource or scope. The user-assigned client UUID is +pinned at startup by `pg_durable.managed_identity_client_id`; an unset or empty +value disables MI without contacting a provider. Endpoint-level `client_id` is +rejected. There is no implicit system-assigned or provider-default identity. +Ordinary headers and +secret bindings cannot override the endpoint's authorization, including before +its token has been fetched. See [Managed Identity](../USER_GUIDE.md#managed-identity) +for the exact host/resource mapping. + +After privilege, catalog, destination and binding checks, and after releasing the +catalog connection, the HTTP activity obtains a token from its configured provider. +The dedicated client may reach IMDS without weakening the request client's IP or +domain controls. Both the provider URL and the pinned identity are protected +startup settings, not workflow data; endpoint administration does not grant +configuration authority. The client uses the public IMDS protocol, disables redirects and proxies, +and bounds response size and acquisition time. There is no provider fallback. +The provider must honor the configured identity and be trusted to supply its +tokens. The selected identity must be approved for customer-configured requests; +pinning it does not itself isolate customer identities from service identities +available elsewhere on the host. These controls protect against SQL-level +redirection, not code already able to access host credentials or the PostgreSQL +process's memory. + +Tokens are validated, marked sensitive and cached only inside the worker, by +resource for its pinned identity. Concurrent misses for the same key share one fetch; +tokens within two minutes of expiry are not reused, and refresh failures never +serve an old token. Provider responses, parse details and credential headers are +not included in activity errors, logs or results. No separate durable token-fetch +activity is scheduled. Responses from the destination retain the existing +response policy, including its limitations around echoed secrets. +Changing the client ID and restarting clears cached tokens and changes subsequent +requests, including pending workflows and retries, without changing recorded +activity results. + +--- + ## 4. Layer 1: IP Blocklist (SSRF protection) ### 4.1 Blocked IPv4 ranges diff --git a/docs/spec-security-model.md b/docs/spec-security-model.md index 6abae9d5..08fb96ed 100644 --- a/docs/spec-security-model.md +++ b/docs/spec-security-model.md @@ -434,13 +434,15 @@ pg_durable supports workflow variables via `df.setvar()/df.getvar()/df.unsetvar( - `df.secret(server text, key text)` constructs a JSONB descriptor with `server` and `key`, interpreted only in explicit `secret_bindings.headers`, `.query` or `.form` slots supplied through `df.with_http_options`. It never returns a credential. Header descriptors may include a literal `prefix`. - Neither helper reads endpoint configuration or credentials. Ordinary data is not searched for descriptors or secret markers. -**Catalog options**: Servers require `auth_scheme`; `base_url` may be omitted only with `auth_scheme 'none'` for named-secret storage. A supplied URL retains all validation requirements, and HTTP endpoint execution always requires one. `header_name` is required only for header authentication. Schemes are `none` (no mapping required for endpoint authentication), `bearer` (`token` mapping option), `header` (`header_value`), and `query` (`query_string`). Named bindings read individual `"secret."` mapping options, allowing native per-key addition, rotation and removal without rewriting other credentials. Values are opaque text, not JSON. The prefix is accepted only in user mappings and confers no endpoint-authentication or ambient-identity behavior; server options remain closed. See [Endpoint Credential Catalog](../USER_GUIDE.md#endpoint-credential-catalog). FDW creation authority uses a native grant separate from `df.grant_usage`. +**Catalog options**: Servers require `auth_scheme`; `base_url` may be omitted only with `auth_scheme 'none'` for named-secret storage. A supplied URL retains all validation requirements, and HTTP endpoint execution always requires one. `header_name` is required only for header authentication. Schemes are `none` (no mapping required for endpoint authentication), `bearer` (`token` mapping option), `header` (`header_value`), `query` (`query_string`), and `managed-identity` (activity-local token acquisition without a mapping). Named bindings read individual `"secret."` mapping options, allowing native per-key addition, rotation and removal without rewriting other credentials. Values are opaque text, not JSON. The prefix is accepted only in user mappings and confers no endpoint-authentication or ambient-identity behavior; server options remain closed. See [Endpoint Credential Catalog](../USER_GUIDE.md#endpoint-credential-catalog). FDW creation authority uses a native grant separate from `df.grant_usage`. + +**Managed identity**: Configuration requires `EXECUTE` on the extension-owned `df.managed_identity_admin()` capability in addition to native FDW and ownership permissions. Execution rechecks the endpoint owner's effective capability, including inherited grants, and fails closed if the capability function is missing or not extension-owned. `PUBLIC` and the ordinary grant/revoke helpers do not receive or manage this capability. The protected startup GUC `pg_durable.managed_identity_client_id` pins one nonzero user-assigned client UUID; empty disables MI without selecting an ambient identity. Endpoint `client_id` options are rejected. The token resource is fixed by a supported HTTPS destination on port 443. Token acquisition follows permission, destination and binding checks, releases the catalog slot first, and uses a separate bounded IMDS-compatible client. Resource cache entries are in-memory only for the pinned identity, without stale-token fallback. Changing the identity requires restart and affects subsequent attempts. Provider isolation remains a deployment responsibility. See [Managed identity security](http-security.md#38-managed-identity). **Body handling**: Explicit form-urlencoded fields are supported; general templates, JSON-body insertion and secret-valued multipart parts are deferred. `form_fields` contains literal strings and `secret_bindings.form` contains references. Both sets are serialized inside the activity without interpreting markers, descriptors or workflow placeholders in ordinary values. Whole-body marker scanning remains unsafe even when opted in, because untrusted data concatenated during graph construction is indistinguishable from intentional references. **Resolution and authorization**: - The activity derives the required HTTP function from the actual destination/body mode and re-checks `EXECUTE`, then checks `USAGE` on every referenced server for `submitted_by`. Hand-crafted node JSON must pass the same checks as helper-produced requests. The grant/revoke helpers cover all HTTP variants; binding configuration itself must come from trusted workflow code, not untrusted request data. -- Credential lookup uses `pg_user_mappings` over `connect_as_user(submitted_by)`, never the worker's privileged pool. Missing servers, mappings, options or masked values fail explicitly; there is no fallback to another role's mapping or an unauthenticated request. +- Static credential lookup uses `pg_user_mappings` over `connect_as_user(submitted_by)`, never the worker's privileged pool. Missing servers, mappings, options or masked values fail explicitly; there is no fallback to another role's mapping or an unauthenticated request. - Catalogs live in the trusted origin installation, independently of the workflow's SQL target or HTTP JSON fields. One read-only consistent snapshot supplies endpoint configuration and all named bindings for an attempt. Satellite catalog connections check database OID and installation UUID after admission. The caller connection uses the shared SQL/catalog admission budget and closes before HTTP I/O; subsequent attempts start fresh snapshots. - Immediately before send, a short metadata transaction revalidates source identity and HTTP `EXECUTE`. This catches revocation/removal during catalog waits without mixing secret snapshots. It does not atomically couple a remote send to source DDL or retract an accepted request. - Compose the destination without allowing a path or substituted value to replace the server's authority. Apply scheme, allow-list and SSRF checks to the actual destination, and keep redirects disabled. Secret substitutions in URL components require context-appropriate encoding. @@ -722,7 +724,7 @@ Per-user scoping of `df.vars` (adding an `owner` column + RLS) is deferred to a #### Endpoint Catalog Objects -Per-role HTTP credentials use native user mappings, not a `df.secrets` table. The extension installs `pg_durable_fdw`, its option validator and reference helpers; endpoint owners manage foreign servers and user mappings using native PostgreSQL DDL. The validator must ship with the wrapper and reject unsupported options and authentication schemes. Managed identity must remain unavailable until its authorization controls are implemented. +Per-role HTTP credentials use native user mappings, not a `df.secrets` table. The extension installs `pg_durable_fdw`, its option validator and reference helpers; endpoint owners manage foreign servers and user mappings using native PostgreSQL DDL. The validator rejects unsupported options and authentication schemes. Managed-identity configuration requires an explicitly delegated capability (or superuser authority) and does not store tokens in the catalog. Credential values remain plaintext in catalogs, WAL and backups. Superuser dumps include user-mapping credentials; dumps by less privileged roles can omit mapping options. Document and test backup/restore behavior. Dropping the extension with `CASCADE` can remove dependent endpoints and mappings. Supplying literal credentials in mapping DDL can also expose them through statement logging. @@ -732,6 +734,7 @@ Credential values remain plaintext in catalogs, WAL and backups. Superuser dumps - The new binary must work against all previous schemas in the same major version without requiring `ALTER EXTENSION UPDATE`. Detect unavailable endpoint objects when a reference is used and fail clearly; existing raw-URL workflows must continue working on older schemas. - Preserve the serialized activity inputs and durable operation sequence for existing workflows. Add endpoint handling without changing the representation of legacy HTTP nodes. - Extend upgrade comparisons to cover FDW/validator definitions and grants, not just objects in the `df` schema. Include representative customer servers and mappings in upgrade tests. +- Managed-identity authentication in 0.2.10 adds `df.managed_identity_admin() RETURNS void` with PUBLIC execution revoked in fresh-install and 0.2.9-to-0.2.10 upgrade SQL. Capability lookup uses native catalogs in the trusted origin installation and denies MI authorization if the object is absent or not extension-owned; ordinary HTTP on older schemas is unaffected. Capability grants are database-local, while the client-ID and provider GUCs apply to the shared worker. Existing durable inputs and operation ordering are unchanged; the provider is consulted only during activity execution, never replay. ### 8.2 Function Permissions (Extension Installation) diff --git a/docs/upgrade-testing.md b/docs/upgrade-testing.md index 75acdc5a..fb08936b 100644 --- a/docs/upgrade-testing.md +++ b/docs/upgrade-testing.md @@ -351,6 +351,41 @@ gate, so they never need to be added to the exclude list. Each schema-changing PR should add a section here documenting what changed, what the upgrade script handles, and any backward compatibility considerations. +### v0.2.9 → v0.2.10 + +#### Managed identity endpoints + +- **Runtime behavior:** `auth_scheme 'managed-identity'` uses the identity pinned + by the protected startup GUC `pg_durable.managed_identity_client_id`. Empty + disables MI without selecting a default identity. Endpoint `client_id` options + are rejected; resources remain derived from destinations. +- **Guarantee A/B2 considerations:** Fresh-install SQL and + [0.2.9 to 0.2.10](../sql/pg_durable--0.2.9--0.2.10.sql) add + `df.managed_identity_admin() RETURNS void` and revoke PUBLIC execution. + The validator requires its effective `EXECUTE` privilege for configuration; + activities check the owner's capability plus caller permissions and destination + policy on each attempt. The B2 probe verifies default denial and explicit + non-superuser delegation. Existing HTTP signatures and grants are unchanged. + The shipped 0.2.8-to-0.2.9 migration remains byte-identical. + Tokens remain in worker memory, never catalog mappings or durable results. +- **Multi-database installations:** Upgrade each installation that will use MI. + Capability and endpoint checks use the trusted origin database, not the SQL + target or a caller-supplied database field. Provider and client-ID GUCs apply to + the shared worker and are independent of local extension schema versions. +- **Guarantee B1 considerations:** Existing workflows still run on supported + schemas from 0.2.2 through 0.2.9 without an extension update. Token-client + construction performs no network access; workflows without MI endpoints never + contact a token provider. Capability lookup checks native catalog identity and + extension membership and fails closed when the capability is absent. B1 checks + its absence separately from HTTP helpers that already shipped in 0.2.9. + No endpoint data conversion is required. +- **Replay compatibility:** Activity names, operation order and serialized inputs + are unchanged. Endpoint configuration and authentication are resolved only when + an HTTP activity executes; recorded results replay without acquiring a token. + Changing the configured identity and restarting affects pending requests and + retries, not recorded completions. The N-1 lifecycle uses the released 0.2.9 + binary as its predecessor. + ### v0.2.8 → v0.2.9 The post-tag changes in #379, #388, #389, #390, and #380 belong to the 0.2.9 diff --git a/scripts/test-e2e-docker.sh b/scripts/test-e2e-docker.sh index 18873c29..e6c60fc4 100755 --- a/scripts/test-e2e-docker.sh +++ b/scripts/test-e2e-docker.sh @@ -57,7 +57,8 @@ SKIP_TESTS=( "83_multi_database_same_oid" "84_multi_database_metadata_fence" "86_multi_database_untrusted_origin" - # Require http-allow-all and the local runner's TLS mock and environment. + # Require unrestricted HTTP and the local runner's TLS mock and environment. + "75_managed_identity" "80_multi_database_http_origin" "85_multi_database_http_admission" ) diff --git a/scripts/test-e2e-local.sh b/scripts/test-e2e-local.sh index 9b7672e1..8378b770 100755 --- a/scripts/test-e2e-local.sh +++ b/scripts/test-e2e-local.sh @@ -28,6 +28,7 @@ # ./scripts/test-e2e-local.sh 00_requires_shared_preload # ./scripts/test-e2e-local.sh 45_connection_limit_timeout # ./scripts/test-e2e-local.sh http_allowed_domains +# ./scripts/test-e2e-local.sh managed_identity # ./scripts/test-e2e-local.sh --http-disabled 47_http_dsl_disabled # ./scripts/test-e2e-local.sh --http-allow-all # ./scripts/test-e2e-local.sh --include-shapes gen- @@ -57,6 +58,11 @@ SETUP_PLAYGROUND_APPLIED=false E2E_ROLE_ENSURED=false VERSION_SHOWN=false PHASE_LOG_MARK=0 +IDENTITY_MOCK_PROCESS="" +IDENTITY_MOCK_DIR="" +IDENTITY_TOKEN_ENDPOINT="" +IDENTITY_PROXY="" +IDENTITY_CERTIFICATE="" declare -a REQUESTED_PHASES=() declare -a MATCHED_TESTS=() @@ -94,6 +100,7 @@ ALL_PHASES=( "http-empty-domains" "http-disabled" "http-allow-all" + "managed-identity" ) PGRX_HOME="$HOME/.pgrx" @@ -189,6 +196,9 @@ phase_label() { http-allow-all) echo "HTTP unrestricted startup policy" ;; + managed-identity) + echo "managed identity (local token and TLS mocks)" + ;; *) echo "$1" ;; @@ -239,6 +249,9 @@ phase_for_test() { 48_http_allow_all|80_multi_database_http_origin|85_multi_database_http_admission) echo "http-allow-all" ;; + 75_managed_identity) + echo "managed-identity" + ;; *) echo "standard" ;; @@ -355,9 +368,11 @@ cleanup() { if [ "$KEEP_RUNNING" = false ]; then stop_server + stop_identity_mock return fi + stop_identity_mock echo "" echo -e "${GREEN}PostgreSQL left running on port $PG_PORT${NC}" echo "Connect: $PSQL -h localhost -p $PG_PORT -d $PG_DB" @@ -367,6 +382,33 @@ cleanup() { trap cleanup EXIT +stop_identity_mock() { + if [ -n "$IDENTITY_MOCK_PROCESS" ]; then + kill "$IDENTITY_MOCK_PROCESS" 2>/dev/null || true + wait "$IDENTITY_MOCK_PROCESS" 2>/dev/null || true + IDENTITY_MOCK_PROCESS="" + fi + if [ -n "$IDENTITY_MOCK_DIR" ]; then + rm -rf -- "$IDENTITY_MOCK_DIR" + IDENTITY_MOCK_DIR="" + remove_conf_key "pg_durable.managed_identity_endpoint" + remove_conf_key "pg_durable.managed_identity_client_id" + fi + IDENTITY_TOKEN_ENDPOINT="" + IDENTITY_PROXY="" + IDENTITY_CERTIFICATE="" +} + +start_identity_mock() { + IDENTITY_MOCK_DIR=$(mktemp -d "${TMPDIR:-/tmp}/pg-durable-mi.XXXXXXXX") + coproc IDENTITY_MOCK { exec python3 "$PROJECT_DIR/tests/e2e/managed_identity_mock.py" "$IDENTITY_MOCK_DIR"; } + IDENTITY_MOCK_PROCESS=$IDENTITY_MOCK_PID + if ! IFS=$'\t' read -r -t 15 IDENTITY_TOKEN_ENDPOINT IDENTITY_PROXY IDENTITY_CERTIFICATE <&"${IDENTITY_MOCK[0]}"; then + echo "TEST FAILED: managed identity mock did not start" + exit 1 + fi +} + remove_conf_key() { local key="$1" local escaped_key="${key//./\\.}" @@ -405,7 +447,13 @@ wait_for_server() { restart_server() { stop_server echo -e "${YELLOW}Starting PostgreSQL...${NC}" - if [ -n "${SERVER_PGHOST:-}" ]; then + if [ -n "$IDENTITY_TOKEN_ENDPOINT" ]; then + HTTP_PROXY="$IDENTITY_PROXY" http_proxy="$IDENTITY_PROXY" \ + HTTPS_PROXY="$IDENTITY_PROXY" https_proxy="$IDENTITY_PROXY" \ + ALL_PROXY="" all_proxy="" NO_PROXY="" no_proxy="" \ + SSL_CERT_FILE="$IDENTITY_CERTIFICATE" \ + "$PG_CTL" -D "$DATA_DIR" -l "$LOG_FILE" start >/dev/null 2>&1 + elif [ -n "${SERVER_PGHOST:-}" ]; then PGHOST="$SERVER_PGHOST" "$PG_CTL" -D "$DATA_DIR" -l "$LOG_FILE" start >/dev/null 2>&1 else "$PG_CTL" -D "$DATA_DIR" -l "$LOG_FILE" start >/dev/null 2>&1 @@ -575,6 +623,8 @@ configure_phase() { remove_conf_key "pg_durable.http_allowed_domains" remove_conf_key "pg_durable.http_security" set_conf_line "pg_durable.http_allowed_domains" "'$PG_DURABLE_TEST_HTTP_DOMAINS'" + remove_conf_key "pg_durable.managed_identity_endpoint" + remove_conf_key "pg_durable.managed_identity_client_id" # Match scripts/pg-common.sh so the shared pgrx cluster keeps a usable socket # directory for `make installcheck` after an E2E run. set_conf_line "unix_socket_directories" "'$PGRX_HOME'" @@ -673,13 +723,22 @@ configure_phase() { set_conf_line "pg_durable.reconcile_interval" "0" fi ;; + managed-identity) + start_identity_mock + set_conf_line "shared_preload_libraries" "'pg_durable'" + set_conf_line "pg_durable.worker_role" "'postgres'" + set_conf_line "pg_durable.database" "'postgres'" + set_conf_line "pg_durable.enable_superuser_instances" "on" + set_conf_line "pg_durable.managed_identity_endpoint" "'$IDENTITY_TOKEN_ENDPOINT'" + set_conf_line "pg_durable.managed_identity_client_id" "'11111111-1111-1111-1111-111111111111'" + ;; esac case "$phase" in http-disabled) set_conf_line "pg_durable.http_security" "'disabled'" ;; - http-allow-all) + http-allow-all|managed-identity) set_conf_line "pg_durable.http_security" "'unrestricted'" ;; esac @@ -687,6 +746,7 @@ configure_phase() { prepare_phase() { local phase="$1" + stop_identity_mock configure_phase "$phase" @@ -707,6 +767,11 @@ prepare_phase() { echo "Checking invalid HTTP startup setting rejection..." assert_http_startup_rejected "pg_durable.http_allowed_domains" "example.com,https://api.github.com" || exit 1 assert_http_startup_rejected "pg_durable.http_security" "disable" || exit 1 + elif [ "$phase" = "managed-identity" ]; then + echo "Checking unsafe managed identity provider startup rejection..." + assert_http_startup_rejected "pg_durable.managed_identity_endpoint" "http://identity.example/token" || exit 1 + assert_http_startup_rejected "pg_durable.managed_identity_client_id" "not-a-uuid" || exit 1 + assert_http_startup_rejected "pg_durable.managed_identity_client_id" "00000000-0000-0000-0000-000000000000" || exit 1 fi if [ -f "$LOG_FILE" ]; then @@ -759,7 +824,7 @@ prepare_phase() { reconcile|maintenance-backlog) wait_for_worker_ready ;; - http-custom-domains|http-empty-domains|http-disabled|http-allow-all) + http-custom-domains|http-empty-domains|http-disabled|http-allow-all|managed-identity) ensure_e2e_role wait_for_worker_ready ;; @@ -1051,6 +1116,17 @@ run_phase() { fi done + if [ "$phase" = "managed-identity" ]; then + printf " %-45s ... " "managed_identity_log_redaction" + if tail -n "+$((PHASE_LOG_MARK + 1))" "$LOG_FILE" | grep -E 'MI_PRIVATE_(SYSTEM_TOKEN|USER_TOKEN|PROVIDER_ERROR)' >/dev/null; then + echo -e "${RED}FAIL${NC}" + phase_failed=$((phase_failed + 1)) + else + echo -e "${GREEN}PASS${NC}" + phase_passed=$((phase_passed + 1)) + fi + fi + if [ "$phase" = "standard" ] && [ -z "$TEST_FILTER" ]; then printf " %-45s ... " "application_name_connection_logs" if assert_application_names_logged; then diff --git a/scripts/test-upgrade.sh b/scripts/test-upgrade.sh index a8941bc3..7adf4c6c 100755 --- a/scripts/test-upgrade.sh +++ b/scripts/test-upgrade.sh @@ -856,6 +856,10 @@ test_b1_http_options_absent() { assert_sql_equals "SELECT to_regprocedure('df.with_http_options(text,jsonb)') IS NULL;" "t" } +test_b1_managed_identity_admin_absent() { + assert_sql_equals "SELECT to_regprocedure('df.managed_identity_admin()') IS NULL;" "t" +} + test_b1_dsl_chain() { assert_sql_contains "SELECT df.sql('SELECT 1') ~> df.sql('SELECT 2');" '"node_type":"THEN"' } @@ -1117,7 +1121,10 @@ else run_test "B1 [v${B1_VERSION}]: df.loop(body, condition)" test_b1_conditional_loop run_test "B1 [v${B1_VERSION}]: df.http() construction" test_b1_http_construction if ! version_ge "$B1_VERSION" "0.2.9"; then - run_test "B1 [v${B1_VERSION}]: new HTTP options helper remains absent" test_b1_http_options_absent + run_test "B1 [v${B1_VERSION}]: HTTP options helper remains absent" test_b1_http_options_absent + fi + if ! version_ge "$B1_VERSION" "0.2.10"; then + run_test "B1 [v${B1_VERSION}]: MI capability remains absent" test_b1_managed_identity_admin_absent fi run_test "B1 [v${B1_VERSION}]: DSL chain (~>)" test_b1_dsl_chain run_test "B1 [v${B1_VERSION}]: conditional operators (?>/!>)" test_b1_conditional_operators @@ -1334,6 +1341,7 @@ test_b2_endpoint_catalog_after_upgrade() { run_sql_capture "CREATE ROLE durable_b2_endpoint_probe LOGIN; SELECT df.grant_usage('durable_b2_endpoint_probe');" >/dev/null || return 1 assert_sql_equals "SELECT pg_catalog.has_foreign_data_wrapper_privilege('durable_b2_endpoint_probe', 'pg_durable_fdw', 'USAGE');" "f" || return 1 + assert_sql_equals "SELECT pg_catalog.has_function_privilege('durable_b2_endpoint_probe', 'df.managed_identity_admin()', 'EXECUTE');" "f" || return 1 run_sql_capture "GRANT USAGE ON FOREIGN DATA WRAPPER pg_durable_fdw TO durable_b2_endpoint_probe; SET ROLE durable_b2_endpoint_probe; CREATE SERVER durable_b2_endpoint FOREIGN DATA WRAPPER pg_durable_fdw @@ -1359,7 +1367,16 @@ test_b2_endpoint_catalog_after_upgrade() { jsonb_build_object('secret_bindings', jsonb_build_object('form', jsonb_build_object('password', df.secret('durable_b2_endpoint', 'key'))), 'form_fields', jsonb_build_object('ordinary', 'literal')) )::jsonb->>'query')::jsonb->'secret_bindings'->'form'->'password'->>'key') = 'key';" "t" || return 1 - run_sql_capture "DROP SERVER durable_b2_endpoint, durable_b2_secrets CASCADE; + run_sql_capture "GRANT EXECUTE ON FUNCTION df.managed_identity_admin() TO durable_b2_endpoint_probe; + SET ROLE durable_b2_endpoint_probe; + SELECT df.managed_identity_admin(); + CREATE SERVER durable_b2_identity FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://account.blob.core.windows.net', auth_scheme 'managed-identity'); + ALTER SERVER durable_b2_identity OPTIONS (SET base_url 'https://other.blob.core.windows.net'); + RESET ROLE; + REVOKE EXECUTE ON FUNCTION df.managed_identity_admin() FROM durable_b2_endpoint_probe;" >/dev/null || return 1 + assert_sql_equals "SELECT pg_catalog.has_function_privilege('durable_b2_endpoint_probe', 'df.managed_identity_admin()', 'EXECUTE');" "f" || return 1 + run_sql_capture "DROP SERVER durable_b2_endpoint, durable_b2_secrets, durable_b2_identity CASCADE; DROP OWNED BY durable_b2_endpoint_probe; DROP ROLE durable_b2_endpoint_probe;" >/dev/null } diff --git a/sql/pg_durable--0.2.9--0.2.10.sql b/sql/pg_durable--0.2.9--0.2.10.sql new file mode 100644 index 00000000..39654efe --- /dev/null +++ b/sql/pg_durable--0.2.9--0.2.10.sql @@ -0,0 +1,8 @@ +CREATE FUNCTION df.managed_identity_admin() +RETURNS pg_catalog.void +LANGUAGE c STRICT +AS 'MODULE_PATHNAME', 'managed_identity_admin_wrapper'; + +REVOKE ALL ON FUNCTION df.managed_identity_admin() FROM PUBLIC; +COMMENT ON FUNCTION df.managed_identity_admin() IS + 'EXECUTE authorizes managed identity endpoint administration. Calling this function does not acquire tokens.'; \ No newline at end of file diff --git a/src/activities/execute_http.rs b/src/activities/execute_http.rs index b8b02174..8b7d6105 100644 --- a/src/activities/execute_http.rs +++ b/src/activities/execute_http.rs @@ -153,6 +153,7 @@ pub(crate) async fn execute( route: &mut crate::origin::Route, semaphore: Arc, policy: Arc, + identity_client: Arc, config_json: String, ) -> Result { let config: HttpConfig = @@ -256,6 +257,7 @@ pub(crate) async fn execute( .resolve(&mut catalog, &mut prepared, config.headers.as_ref()) .await?; catalog.close().await?; + prepared.authorize(&identity_client).await?; let safe_url = if config .secret_options .secret_bindings diff --git a/src/activities/execute_multipart.rs b/src/activities/execute_multipart.rs index 01b7518c..d550f7fe 100644 --- a/src/activities/execute_multipart.rs +++ b/src/activities/execute_multipart.rs @@ -11,7 +11,7 @@ //! `reqwest::multipart::Form` built from base64-encoded parts) and the //! privilege target (`df.http_multipart` instead of `df.http`). //! -//! Startup settings controlling outbound HTTP(S) are the same as for df.http — +//! The startup `pg_durable.http_security` policy is the same as for df.http — //! see docs/http-security.md for the full security model. use base64::Engine as _; @@ -107,6 +107,7 @@ pub(crate) async fn execute( route: &mut crate::origin::Route, semaphore: Arc, policy: Arc, + identity_client: Arc, config_json: String, ) -> Result { let config: MultipartConfig = serde_json::from_str(&config_json) @@ -192,6 +193,7 @@ pub(crate) async fn execute( .resolve(&mut catalog, &mut prepared, config.headers.as_ref()) .await?; catalog.close().await?; + prepared.authorize(&identity_client).await?; let safe_url = if config .secret_options .secret_bindings diff --git a/src/endpoints.rs b/src/endpoints.rs index a12ef81d..dd84b819 100644 --- a/src/endpoints.rs +++ b/src/endpoints.rs @@ -8,6 +8,25 @@ use url::Url; pub const FDW_NAME: &str = "pg_durable_fdw"; +const MANAGED_IDENTITY_ADMIN_CHECK: &str = + "SELECT EXISTS ( + SELECT 1 FROM pg_catalog.pg_depend AS dependency + JOIN pg_catalog.pg_extension AS extension + ON extension.oid OPERATOR(pg_catalog.=) dependency.refobjid + JOIN pg_catalog.pg_proc AS capability + ON capability.oid OPERATOR(pg_catalog.=) dependency.objid + JOIN pg_catalog.pg_namespace AS namespace + ON namespace.oid OPERATOR(pg_catalog.=) capability.pronamespace + WHERE dependency.classid OPERATOR(pg_catalog.=) 'pg_catalog.pg_proc'::pg_catalog.regclass + AND namespace.nspname OPERATOR(pg_catalog.=) 'df' + AND capability.proname OPERATOR(pg_catalog.=) 'managed_identity_admin' + AND capability.pronargs OPERATOR(pg_catalog.=) 0 + AND dependency.refclassid OPERATOR(pg_catalog.=) 'pg_catalog.pg_extension'::pg_catalog.regclass + AND dependency.deptype OPERATOR(pg_catalog.=) 'e' + AND extension.extname OPERATOR(pg_catalog.=) 'pg_durable' + AND pg_catalog.has_function_privilege($1::pg_catalog.oid, dependency.objid, 'EXECUTE') + )"; + pgrx::extension_sql!( "CREATE TYPE df.http_endpoint AS (server pg_catalog.text, path pg_catalog.text);", name = "create_endpoint_type", @@ -128,6 +147,26 @@ fn compose_endpoint_url(base: &Url, path: &str) -> Result { pub struct EndpointRequest { pub url: Url, pub credential_header: Option<(HeaderName, HeaderValue)>, + pub managed_identity: Option, +} + +impl EndpointRequest { + pub fn credential_header_name(&self) -> Option<&HeaderName> { + self.credential_header + .as_ref() + .map(|(name, _)| name) + .or_else(|| self.managed_identity.as_ref().map(|_| &AUTHORIZATION)) + } + + pub async fn authorize( + &mut self, + client: &crate::managed_identity::TokenClient, + ) -> Result<(), String> { + if let Some(identity) = &self.managed_identity { + self.credential_header = Some((AUTHORIZATION, client.authorization(identity).await?)); + } + Ok(()) + } } fn prepare_endpoint_request( @@ -137,7 +176,7 @@ fn prepare_endpoint_request( ) -> Result { let mut url = compose_endpoint_url(&endpoint.base_url, path)?; let credential_name = match &endpoint.auth { - EndpointAuth::Bearer(_) => Some(&AUTHORIZATION), + EndpointAuth::Bearer(_) | EndpointAuth::ManagedIdentity(_) => Some(&AUTHORIZATION), EndpointAuth::Header { name, .. } => Some(name), _ => None, }; @@ -153,10 +192,15 @@ fn prepare_endpoint_request( } } } + let mut managed_identity = None; let credential_header = match endpoint.auth { EndpointAuth::None => None, EndpointAuth::Bearer(value) => Some((AUTHORIZATION, value)), EndpointAuth::Header { name, value } => Some((name, value)), + EndpointAuth::ManagedIdentity(identity) => { + managed_identity = Some(identity); + None + } EndpointAuth::Query(query) => { let credential_url = Url::parse(&format!("https://endpoint.invalid/?{query}")) .map_err(|_| "Invalid endpoint credential query")?; @@ -181,6 +225,7 @@ fn prepare_endpoint_request( Ok(EndpointRequest { url, credential_header, + managed_identity, }) } @@ -199,6 +244,7 @@ pub async fn prepare_request( None => Ok(EndpointRequest { url: crate::ssrf::parse_request_url(url)?, credential_header: None, + managed_identity: None, }), } } @@ -209,6 +255,7 @@ pub enum AuthScheme { Bearer, Header(HeaderName), Query, + ManagedIdentity(crate::managed_identity::TokenResource), } #[derive(Clone)] @@ -302,12 +349,14 @@ impl EndpointConfig { } AuthScheme::Header(name) } - "managed-identity" => { - return Err("Managed identity is not supported in this version".into()) - } + "managed-identity" => AuthScheme::ManagedIdentity( + crate::managed_identity::TokenResource::for_endpoint( + base_url.as_ref().ok_or("Managed identity requires endpoint base_url")?, + )?, + ), _ => { return Err( - "Unsupported endpoint auth_scheme; allowed: none, bearer, header, query".into(), + "Unsupported endpoint auth_scheme; allowed: none, bearer, header, query, managed-identity".into(), ) } }; @@ -358,10 +407,35 @@ fn validate_mapping_options(options: &[String]) -> Result<(), String> { Ok(()) } +#[pg_extern(schema = "df")] +pub fn managed_identity_admin() {} + +pgrx::extension_sql!( + r#" +REVOKE ALL ON FUNCTION df.managed_identity_admin() FROM PUBLIC; +COMMENT ON FUNCTION df.managed_identity_admin() IS + 'EXECUTE authorizes managed identity endpoint administration. Calling this function does not acquire tokens.'; +"#, + name = "managed_identity_admin_acl", + requires = [managed_identity_admin] +); + #[pg_extern(schema = "df")] pub fn endpoint_option_validator(options: Vec, catalog: pg_sys::Oid) { let result = if catalog == pg_sys::ForeignServerRelationId { - EndpointConfig::from_options(&options).map(|_| ()) + EndpointConfig::from_options(&options).and_then(|config| { + if matches!(config.auth_scheme, AuthScheme::ManagedIdentity(_)) { + let permitted = Spi::get_one_with_args::( + MANAGED_IDENTITY_ADMIN_CHECK, + &[unsafe { pg_sys::GetUserId() }.into()], + ).map_err(|_| "Managed identity administration privilege check failed")? + .unwrap_or(false); + if !permitted { + return Err("Permission denied: EXECUTE on df.managed_identity_admin() is required to configure managed identity endpoints".into()); + } + } + Ok(()) + }) } else if catalog == pg_sys::UserMappingRelationId { validate_mapping_options(&options) } else if catalog == pg_sys::ForeignDataWrapperRelationId { @@ -396,6 +470,7 @@ pub enum EndpointAuth { value: HeaderValue, }, Query(String), + ManagedIdentity(crate::managed_identity::TokenResource), } pub struct ResolvedEndpoint { @@ -408,6 +483,7 @@ fn resolve_auth(config: AuthScheme, mapping: &[String]) -> Result Ok(EndpointAuth::None), + AuthScheme::ManagedIdentity(identity) => Ok(EndpointAuth::ManagedIdentity(identity)), AuthScheme::Bearer => { let mut value = HeaderValue::from_str(&format!("Bearer {}", required(&mapping, "token")?)) @@ -430,6 +506,15 @@ fn resolve_auth(config: AuthScheme, mapping: &[String]) -> Result>, +} + struct CatalogServer { oid: i64, config: EndpointConfig, @@ -563,22 +648,28 @@ impl<'a> EndpointCatalog<'a> { .connection() .await .map_err(|error| format!("Endpoint server {server:?}: {error}"))?; - let endpoint: Option<(i64, bool, bool, Option>)> = sqlx::query_as( - "SELECT server.oid::pg_catalog.int8, - wrapper.fdwname OPERATOR(pg_catalog.=) $2, - pg_catalog.has_server_privilege(server.oid, 'USAGE'), - server.srvoptions + let endpoint = sqlx::query_as::<_, CatalogServerRow>( + "SELECT server.oid::pg_catalog.int8 AS oid, + wrapper.fdwname OPERATOR(pg_catalog.=) $2 AS correct_wrapper, + pg_catalog.has_server_privilege(server.oid, 'USAGE') AS permitted, + server.srvowner AS owner_oid, + server.srvoptions AS options FROM pg_catalog.pg_foreign_server AS server JOIN pg_catalog.pg_foreign_data_wrapper AS wrapper ON wrapper.oid OPERATOR(pg_catalog.=) server.srvfdw WHERE server.srvname OPERATOR(pg_catalog.=) $1", ) .bind(server) .bind(FDW_NAME) - .fetch_optional(connection) + .fetch_optional(&mut *connection) .await .map_err(|_| "Endpoint server lookup failed")?; - let (oid, correct_wrapper, permitted, options) = - endpoint.ok_or("Endpoint server does not exist")?; + let CatalogServerRow { + oid, + correct_wrapper, + permitted, + owner_oid, + options, + } = endpoint.ok_or("Endpoint server does not exist")?; if !correct_wrapper { return Err("Endpoint server must use pg_durable_fdw".into()); } @@ -586,6 +677,16 @@ impl<'a> EndpointCatalog<'a> { return Err("Permission denied: endpoint server USAGE is required".into()); } let config = EndpointConfig::from_options(options.as_deref().unwrap_or_default())?; + if matches!(config.auth_scheme, AuthScheme::ManagedIdentity(_)) { + let owner_permitted: bool = sqlx::query_scalar(MANAGED_IDENTITY_ADMIN_CHECK) + .bind(owner_oid) + .fetch_one(connection) + .await + .map_err(|_| "Managed identity endpoint owner privilege check failed")?; + if !owner_permitted { + return Err("Permission denied: managed identity endpoint owner must have EXECUTE on df.managed_identity_admin()".into()); + } + } self.servers.insert( server.to_owned(), CatalogServer { @@ -636,8 +737,11 @@ impl<'a> EndpointCatalog<'a> { "Endpoint server {server:?} has no base_url; it can only be used for named secrets" ) })?; - let auth = if matches!(config.auth_scheme, AuthScheme::None) { - EndpointAuth::None + let auth = if matches!( + config.auth_scheme, + AuthScheme::None | AuthScheme::ManagedIdentity(_) + ) { + resolve_auth(config.auth_scheme, &[])? } else { resolve_auth(config.auth_scheme, self.load_mapping(server).await?)? }; @@ -829,6 +933,65 @@ mod unit_tests { values.iter().map(|value| value.to_string()).collect() } + #[test] + fn endpoint_managed_identity_options_and_headers() { + let config = EndpointConfig::from_options(&options(&[ + "base_url=https://account.blob.core.windows.net/container", + "auth_scheme=managed-identity", + ])) + .unwrap(); + assert!(matches!(config.auth_scheme, AuthScheme::ManagedIdentity(_))); + let endpoint = || ResolvedEndpoint { + base_url: config.base_url.clone().unwrap(), + auth: resolve_auth(config.auth_scheme.clone(), &[]).unwrap(), + }; + for header in ["Authorization", "aUtHoRiZaTiOn", "Host"] { + assert!(prepare_endpoint_request( + endpoint(), + "/blob", + Some(&serde_json::json!({header: "override"})) + ) + .is_err()); + } + let request = prepare_endpoint_request(endpoint(), "/blob", None).unwrap(); + assert!(request.credential_header.is_none()); + assert_eq!(request.credential_header_name(), Some(&AUTHORIZATION)); + assert!(request.managed_identity.is_some()); + for invalid in [ + vec!["auth_scheme=managed-identity"], + vec![ + "base_url=https://account.blob.core.windows.net", + "auth_scheme=managed-identity", + "client_id=12345678-1234-1234-1234-123456789abc", + ], + vec![ + "base_url=https://account.blob.core.windows.net", + "auth_scheme=managed-identity", + "client_id=PRIVATE_VALUE", + ], + vec![ + "base_url=https://account.blob.core.windows.net", + "auth_scheme=managed-identity", + "resource=PRIVATE_VALUE", + ], + vec![ + "base_url=https://account.blob.core.windows.net", + "auth_scheme=managed-identity", + "scope=PRIVATE_VALUE", + ], + vec![ + "base_url=https://account.blob.core.windows.net", + "auth_scheme=none", + "client_id=PRIVATE_VALUE", + ], + ] { + let error = EndpointConfig::from_options(&options(&invalid)) + .err() + .unwrap(); + assert!(!error.contains("PRIVATE_VALUE")); + } + } + #[test] fn endpoint_valid_options() { let secrets_only = EndpointConfig::from_options(&options(&["auth_scheme=none"])).unwrap(); @@ -933,6 +1096,129 @@ mod unit_tests { mod tests { use super::*; + fn endpoint_managed_identity_permissions() { + let admin = Spi::get_one::("SELECT CURRENT_USER::text") + .unwrap() + .unwrap(); + let database = Spi::get_one::("SELECT pg_catalog.current_database()::text") + .unwrap() + .unwrap(); + tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async { + let mut connection = crate::types::connect_as_user(&admin, Some(&database)).await.unwrap(); + sqlx::raw_sql(r#" + DROP ROLE IF EXISTS endpoint_mi_user, endpoint_mi_other, endpoint_mi_owner, endpoint_mi_admin; + CREATE ROLE endpoint_mi_user LOGIN; + CREATE ROLE endpoint_mi_other LOGIN; + CREATE ROLE endpoint_mi_owner SUPERUSER; + CREATE ROLE endpoint_mi_admin; + GRANT USAGE ON FOREIGN DATA WRAPPER pg_durable_fdw TO endpoint_mi_user; + CREATE SERVER endpoint_mi_test FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://account.blob.core.windows.net', auth_scheme 'managed-identity'); + CREATE SERVER endpoint_mi_plain FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://api.github.com', auth_scheme 'none'); + GRANT USAGE ON FOREIGN SERVER endpoint_mi_test TO endpoint_mi_user; + GRANT USAGE ON FOREIGN SERVER endpoint_mi_plain TO endpoint_mi_user; + "#).execute(&mut connection).await.unwrap(); + let permitted: bool = sqlx::query_scalar("SELECT pg_catalog.has_function_privilege('endpoint_mi_user', 'df.managed_identity_admin()', 'EXECUTE')") + .fetch_one(&mut connection).await.unwrap(); + assert!(!permitted); + let mut user = crate::types::connect_as_user("endpoint_mi_user", Some(&database)).await.unwrap(); + let endpoint = resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_test").await.unwrap(); + assert!(matches!(endpoint.auth, EndpointAuth::ManagedIdentity(_))); + assert!(resolve_endpoint("endpoint_mi_other", Some(&database), "endpoint_mi_test").await.err().unwrap().contains("USAGE")); + + let error = sqlx::raw_sql("CREATE SERVER endpoint_mi_owned FOREIGN DATA WRAPPER pg_durable_fdw OPTIONS (base_url 'https://account.blob.core.windows.net', auth_scheme 'managed-identity')") + .execute(&mut user).await.unwrap_err(); + assert!(error.to_string().contains("EXECUTE on df.managed_identity_admin()")); + sqlx::raw_sql("CREATE SERVER endpoint_mi_owned FOREIGN DATA WRAPPER pg_durable_fdw OPTIONS (base_url 'https://account.blob.core.windows.net', auth_scheme 'none')") + .execute(&mut user).await.unwrap(); + let error = sqlx::raw_sql("ALTER SERVER endpoint_mi_owned OPTIONS (SET auth_scheme 'managed-identity')") + .execute(&mut user).await.unwrap_err(); + assert!(error.to_string().contains("EXECUTE on df.managed_identity_admin()")); + + sqlx::raw_sql("ALTER SERVER endpoint_mi_test OWNER TO endpoint_mi_user").execute(&mut connection).await.unwrap(); + let error = resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_test").await.err().unwrap(); + assert!(error.contains("owner must have EXECUTE on df.managed_identity_admin()"), "{error}"); + let error = sqlx::raw_sql("ALTER SERVER endpoint_mi_test OPTIONS (SET base_url 'https://other.blob.core.windows.net')") + .execute(&mut user).await.unwrap_err(); + assert!(error.to_string().contains("EXECUTE on df.managed_identity_admin()")); + + sqlx::raw_sql("GRANT EXECUTE ON FUNCTION df.managed_identity_admin() TO endpoint_mi_user") + .execute(&mut connection).await.unwrap(); + sqlx::raw_sql("ALTER SERVER endpoint_mi_owned OPTIONS (SET auth_scheme 'managed-identity'); + CREATE SERVER endpoint_mi_delegated FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://account.blob.core.windows.net', auth_scheme 'managed-identity')") + .execute(&mut user).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_test").await.is_ok()); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_owned").await.is_ok()); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_delegated").await.is_ok()); + sqlx::raw_sql("REVOKE EXECUTE ON FUNCTION df.managed_identity_admin() FROM endpoint_mi_user") + .execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_delegated").await.err().unwrap().contains("owner must have EXECUTE")); + + sqlx::raw_sql("GRANT EXECUTE ON FUNCTION df.managed_identity_admin() TO endpoint_mi_admin; + GRANT USAGE ON FOREIGN DATA WRAPPER pg_durable_fdw TO endpoint_mi_admin; + GRANT endpoint_mi_admin TO endpoint_mi_user") + .execute(&mut connection).await.unwrap(); + sqlx::raw_sql("ALTER SERVER endpoint_mi_delegated OPTIONS (SET base_url 'https://other.blob.core.windows.net')") + .execute(&mut user).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_delegated").await.is_ok()); + sqlx::raw_sql("REVOKE endpoint_mi_admin FROM endpoint_mi_user; + ALTER ROLE endpoint_mi_user NOINHERIT; + GRANT endpoint_mi_admin TO endpoint_mi_user") + .execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_delegated").await.err().unwrap().contains("owner must have EXECUTE")); + let error = sqlx::raw_sql("ALTER SERVER endpoint_mi_delegated OPTIONS (SET base_url 'https://account.blob.core.windows.net')") + .execute(&mut user).await.unwrap_err(); + assert!(error.to_string().contains("EXECUTE on df.managed_identity_admin()")); + sqlx::raw_sql("SET ROLE endpoint_mi_admin; + CREATE SERVER endpoint_mi_group FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://account.blob.core.windows.net', auth_scheme 'managed-identity'); + RESET ROLE") + .execute(&mut user).await.unwrap(); + sqlx::raw_sql("GRANT USAGE ON FOREIGN SERVER endpoint_mi_group TO endpoint_mi_user") + .execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_group").await.is_ok()); + + sqlx::raw_sql("ALTER EXTENSION pg_durable DROP FUNCTION df.managed_identity_admin()") + .execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_group").await.err().unwrap().contains("owner must have EXECUTE")); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_plain").await.is_ok()); + let error = sqlx::raw_sql("CREATE SERVER endpoint_mi_forged FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://account.blob.core.windows.net', auth_scheme 'managed-identity')") + .execute(&mut connection).await.unwrap_err(); + assert!(error.to_string().contains("EXECUTE on df.managed_identity_admin()")); + sqlx::raw_sql("ALTER EXTENSION pg_durable ADD FUNCTION df.managed_identity_admin(); + ALTER FUNCTION df.managed_identity_admin() RENAME TO endpoint_mi_admin_hidden") + .execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_group").await.err().unwrap().contains("owner must have EXECUTE")); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_plain").await.is_ok()); + sqlx::raw_sql("ALTER FUNCTION df.endpoint_mi_admin_hidden() RENAME TO managed_identity_admin") + .execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_group").await.is_ok()); + sqlx::raw_sql("REVOKE EXECUTE ON FUNCTION df.managed_identity_admin() FROM endpoint_mi_admin") + .execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_group").await.err().unwrap().contains("owner must have EXECUTE")); + + sqlx::raw_sql("ALTER SERVER endpoint_mi_test OWNER TO endpoint_mi_owner; ALTER ROLE endpoint_mi_owner NOSUPERUSER; GRANT USAGE ON FOREIGN SERVER endpoint_mi_test TO endpoint_mi_user") + .execute(&mut connection).await.unwrap(); + let error = resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_test").await.err().unwrap(); + assert!(error.contains("owner must have EXECUTE on df.managed_identity_admin()"), "{error}"); + sqlx::raw_sql("ALTER SERVER endpoint_mi_test OWNER TO CURRENT_USER").execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_test").await.is_ok()); + sqlx::raw_sql("REVOKE USAGE ON FOREIGN SERVER endpoint_mi_test FROM endpoint_mi_user").execute(&mut connection).await.unwrap(); + assert!(resolve_endpoint("endpoint_mi_user", Some(&database), "endpoint_mi_test").await.err().unwrap().contains("USAGE")); + + user.close().await.unwrap(); + sqlx::raw_sql(r#" + DROP SERVER endpoint_mi_test, endpoint_mi_owned, endpoint_mi_delegated, endpoint_mi_group, endpoint_mi_plain; + DROP OWNED BY endpoint_mi_user, endpoint_mi_other, endpoint_mi_owner, endpoint_mi_admin; + DROP ROLE endpoint_mi_user, endpoint_mi_other, endpoint_mi_owner, endpoint_mi_admin; + "#).execute(&mut connection).await.unwrap(); + connection.close().await.unwrap(); + }); + } + async fn resolve_endpoint( submitted_by: &str, database: Option<&str>, @@ -1341,6 +1627,7 @@ mod tests { #[pg_test] fn endpoint_catalog_permissions() { + endpoint_managed_identity_permissions(); let admin = Spi::get_one::("SELECT CURRENT_USER::text") .unwrap() .unwrap(); diff --git a/src/lib.rs b/src/lib.rs index fba8c1ef..a71d2939 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -28,6 +28,12 @@ pub static HTTP_SECURITY: GucSetting = pub static HTTP_ALLOWED_DOMAINS: GucSetting> = GucSetting::>::new(Some(ssrf::DEFAULT_HTTP_ALLOWED_DOMAINS)); +pub static MANAGED_IDENTITY_ENDPOINT: GucSetting> = + GucSetting::>::new(Some(managed_identity::DEFAULT_TOKEN_ENDPOINT)); + +pub static MANAGED_IDENTITY_CLIENT_ID: GucSetting> = + GucSetting::>::new(Some(c"")); + pub static MAX_MANAGEMENT_CONNECTIONS: GucSetting = GucSetting::::new(6); pub static MAX_ORIGIN_CONNECTIONS: GucSetting = GucSetting::::new(12); pub static MAX_DUROXIDE_CONNECTIONS: GucSetting = GucSetting::::new(10); @@ -75,6 +81,7 @@ pub mod client; pub mod dsl; pub mod endpoints; pub mod explain; +pub mod managed_identity; pub mod monitoring; pub mod node_status; pub mod orchestrations; @@ -183,6 +190,28 @@ pub extern "C-unwind" fn _PG_init() { None, None, ); + GucRegistry::define_string_guc_with_hooks( + c"pg_durable.managed_identity_endpoint", + c"IMDS-compatible token endpoint for managed identity authentication", + c"Used only for managed identity endpoints. HTTP is restricted to loopback or the IMDS address; HTTPS providers are also supported. Requires a server restart to change.", + &MANAGED_IDENTITY_ENDPOINT, + GucContext::Postmaster, + GucFlags::SUPERUSER_ONLY, + Some(managed_identity::check_token_endpoint), + None, + None, + ); + GucRegistry::define_string_guc_with_hooks( + c"pg_durable.managed_identity_client_id", + c"User-assigned identity used by all managed identity endpoints", + c"An empty value disables managed identity without affecting other HTTP authentication. Requires a nonzero client UUID and a server restart to change. Does not grant endpoint administration privileges.", + &MANAGED_IDENTITY_CLIENT_ID, + GucContext::Postmaster, + GucFlags::SUPERUSER_ONLY, + Some(managed_identity::check_client_id), + None, + None, + ); } GucRegistry::define_int_guc( diff --git a/src/managed_identity.rs b/src/managed_identity.rs new file mode 100644 index 00000000..f66c6f9f --- /dev/null +++ b/src/managed_identity.rs @@ -0,0 +1,803 @@ +use std::collections::BTreeMap; +use std::sync::{Arc, OnceLock}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use reqwest::header::HeaderValue; +use serde::Deserialize; +use tokio::sync::Mutex; +use url::Url; +use uuid::Uuid; + +pub const DEFAULT_TOKEN_ENDPOINT: &std::ffi::CStr = + c"http://169.254.169.254/metadata/identity/oauth2/token"; +const TOKEN_TIMEOUT: Duration = Duration::from_secs(10); +const REFRESH_MARGIN: Duration = Duration::from_secs(120); +const MAX_TOKEN_RESPONSE_BYTES: usize = 64 * 1024; + +pub fn parse_client_id(value: &str) -> Result, String> { + if value.is_empty() { + return Ok(None); + } + Uuid::parse_str(value) + .ok() + .filter(|client_id| !client_id.is_nil()) + .map(Some) + .ok_or_else(|| "Managed identity client ID must be a nonzero UUID, or empty to disable managed identity".into()) +} + +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub struct TokenResource { + resource: &'static str, +} + +impl TokenResource { + pub fn for_endpoint(url: &Url) -> Result { + if url.scheme() != "https" || url.port_or_known_default() != Some(443) { + return Err("Managed identity destinations must use HTTPS on port 443".into()); + } + let host = url + .host_str() + .ok_or("Managed identity destination has no hostname")?; + let resource = if host == "management.azure.com" { + "https://management.azure.com/" + } else { + [ + (".openai.azure.com", "https://cognitiveservices.azure.com"), + ( + ".cognitiveservices.azure.com", + "https://cognitiveservices.azure.com", + ), + ( + ".services.ai.azure.com", + "https://cognitiveservices.azure.com", + ), + (".blob.core.windows.net", "https://storage.azure.com/"), + (".blob.storage.azure.net", "https://storage.azure.com/"), + (".dfs.core.windows.net", "https://storage.azure.com/"), + (".queue.core.windows.net", "https://storage.azure.com/"), + (".table.core.windows.net", "https://storage.azure.com/"), + (".file.core.windows.net", "https://storage.azure.com/"), + (".vault.azure.net", "https://vault.azure.net"), + ] + .into_iter() + .find(|(suffix, _)| host.len() > suffix.len() && host.ends_with(suffix)) + .map(|(_, resource)| resource) + .ok_or("Managed identity is not supported for this destination hostname")? + }; + Ok(Self { resource }) + } +} + +pub fn validate_token_endpoint(value: &str) -> Result { + let endpoint = Url::parse(value).map_err(|_| "Invalid managed identity token endpoint")?; + let local_http = endpoint.scheme() == "http" + && matches!(endpoint.host(), Some(url::Host::Ipv4(address)) if address.is_loopback() || address.octets() == [169, 254, 169, 254]) + || endpoint.scheme() == "http" + && matches!(endpoint.host(), Some(url::Host::Ipv6(address)) if address.is_loopback()); + if (!local_http && endpoint.scheme() != "https") + || endpoint.host_str().is_none() + || !endpoint.username().is_empty() + || endpoint.password().is_some() + || endpoint.query().is_some() + || endpoint.fragment().is_some() + || value + .chars() + .any(|character| character.is_control() || character.is_whitespace()) + || value.contains(['{', '}', '\\']) + { + return Err("Managed identity token endpoint must be HTTPS, or HTTP on a loopback or IMDS address, without userinfo, query or fragment".into()); + } + Ok(endpoint) +} + +#[pgrx::pg_guard] +pub(crate) unsafe extern "C-unwind" fn check_token_endpoint( + newval: *mut *mut std::ffi::c_char, + _extra: *mut *mut std::ffi::c_void, + _source: pgrx::pg_sys::GucSource::Type, +) -> bool { + let result = if unsafe { (*newval).is_null() } { + Err("Managed identity token endpoint must not be NULL".into()) + } else { + unsafe { std::ffi::CStr::from_ptr(*newval) } + .to_str() + .map_err(|_| "Managed identity token endpoint must be UTF-8".into()) + .and_then(validate_token_endpoint) + }; + check_guc_result("pg_durable.managed_identity_endpoint", result.map(|_| ())) +} + +#[pgrx::pg_guard] +pub(crate) unsafe extern "C-unwind" fn check_client_id( + newval: *mut *mut std::ffi::c_char, + _extra: *mut *mut std::ffi::c_void, + _source: pgrx::pg_sys::GucSource::Type, +) -> bool { + let result = if unsafe { (*newval).is_null() } { + Ok(None) + } else { + unsafe { std::ffi::CStr::from_ptr(*newval) } + .to_str() + .map_err(|_| "Managed identity client ID must be UTF-8".into()) + .and_then(parse_client_id) + }; + check_guc_result("pg_durable.managed_identity_client_id", result.map(|_| ())) +} + +fn check_guc_result(parameter: &str, result: Result<(), String>) -> bool { + match result { + Ok(_) => true, + Err(error) => { + if unsafe { pgrx::pg_sys::process_shared_preload_libraries_in_progress } { + pgrx::error!("invalid value for parameter \"{parameter}\": {error}"); + } + unsafe { + pgrx::pg_sys::GUC_check_errdetail_string = + pgrx::PgMemoryContexts::ErrorContext.pstrdup(&error); + } + false + } + } +} + +#[derive(Deserialize)] +#[serde(untagged)] +enum Seconds { + Number(u64), + Text(String), +} + +impl Seconds { + fn value(self) -> Result { + match self { + Self::Number(value) => Ok(value), + Self::Text(value) => value + .parse() + .map_err(|_| "Invalid managed identity token expiry".into()), + } + } +} + +#[derive(Deserialize)] +struct TokenResponse { + access_token: String, + token_type: String, + resource: String, + expires_on: Option, + expires_in: Option, + not_before: Option, + error: Option, +} + +struct CachedToken { + authorization: HeaderValue, + refresh_at: Instant, + refresh_by: SystemTime, +} + +impl CachedToken { + fn parse( + bytes: &[u8], + identity: &TokenResource, + requested_at: SystemTime, + ) -> Result { + let response: TokenResponse = + serde_json::from_slice(bytes).map_err(|_| "Invalid managed identity token response")?; + if response.error.is_some() + || !response.token_type.eq_ignore_ascii_case("Bearer") + || response.resource != identity.resource + || response.access_token.is_empty() + || !response + .access_token + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"-._~+/=".contains(&byte)) + { + return Err("Invalid managed identity token response".into()); + } + let expires_on = match (response.expires_on, response.expires_in) { + (Some(expiry), _) => UNIX_EPOCH.checked_add(Duration::from_secs(expiry.value()?)), + (None, Some(lifetime)) => { + requested_at.checked_add(Duration::from_secs(lifetime.value()?)) + } + (None, None) => return Err("Managed identity token response has no expiry".into()), + } + .ok_or("Invalid managed identity token expiry")?; + let now = SystemTime::now(); + if let Some(not_before) = response.not_before { + let not_before = UNIX_EPOCH + .checked_add(Duration::from_secs(not_before.value()?)) + .ok_or("Invalid managed identity token validity period")?; + if not_before > now { + return Err("Managed identity token is not yet valid".into()); + } + } + let refresh_by = expires_on + .checked_sub(REFRESH_MARGIN) + .ok_or("Invalid managed identity token expiry")?; + let remaining = refresh_by + .duration_since(now) + .map_err(|_| "Managed identity token is expired or too close to expiry")?; + let refresh_at = Instant::now() + .checked_add(remaining) + .ok_or("Invalid managed identity token expiry")?; + let mut authorization = HeaderValue::from_str(&format!("Bearer {}", response.access_token)) + .map_err(|_| "Invalid managed identity access token")?; + authorization.set_sensitive(true); + Ok(Self { + authorization, + refresh_at, + refresh_by, + }) + } + + fn is_fresh(&self) -> bool { + Instant::now() < self.refresh_at && SystemTime::now() < self.refresh_by + } +} + +pub struct TokenClient { + endpoint: Url, + client_id: Option, + client: OnceLock>, + tokens: Mutex>>>>, +} + +impl TokenClient { + pub fn new(endpoint: &str, client_id: Option) -> Result { + let endpoint = validate_token_endpoint(endpoint)?; + Ok(Self { + endpoint, + client_id, + client: OnceLock::new(), + tokens: Mutex::new(BTreeMap::new()), + }) + } + + pub async fn authorization(&self, identity: &TokenResource) -> Result { + let client_id = self.client_id.ok_or("Managed identity is disabled; configure pg_durable.managed_identity_client_id and restart PostgreSQL")?; + tokio::time::timeout( + TOKEN_TIMEOUT, + self.cached_authorization(identity, client_id), + ) + .await + .map_err(|_| "Managed identity token acquisition timed out")? + } + + async fn cached_authorization( + &self, + identity: &TokenResource, + client_id: Uuid, + ) -> Result { + let entry = { + let mut tokens = self.tokens.lock().await; + tokens + .entry(identity.clone()) + .or_insert_with(|| Arc::new(Mutex::new(None))) + .clone() + }; + let mut token = entry.lock().await; + if let Some(token) = token.as_ref().filter(|token| token.is_fresh()) { + return Ok(token.authorization.clone()); + } + *token = None; + let fetched = self.fetch(identity, client_id).await?; + let authorization = fetched.authorization.clone(); + *token = Some(fetched); + Ok(authorization) + } + + async fn fetch( + &self, + identity: &TokenResource, + client_id: Uuid, + ) -> Result { + let client = self + .client + .get_or_init(|| { + reqwest::Client::builder() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .connect_timeout(Duration::from_secs(2)) + .timeout(TOKEN_TIMEOUT) + .build() + .map_err(|_| "Failed to initialize managed identity token client".into()) + }) + .as_ref() + .map_err(Clone::clone)?; + let requested_at = SystemTime::now(); + let mut url = self.endpoint.clone(); + { + let mut query = url.query_pairs_mut(); + query + .append_pair("api-version", "2018-02-01") + .append_pair("resource", identity.resource) + .append_pair("client_id", &client_id.to_string()); + } + let mut response = client + .get(url) + .header("Metadata", "true") + .send() + .await + .map_err(|_| "Managed identity token provider is unavailable")?; + if response.status() != reqwest::StatusCode::OK { + return Err(format!( + "Managed identity token provider returned HTTP {}", + response.status().as_u16() + )); + } + if response + .content_length() + .is_some_and(|size| size > MAX_TOKEN_RESPONSE_BYTES as u64) + { + return Err("Managed identity token response exceeds the size limit".into()); + } + let mut body = Vec::new(); + while let Some(chunk) = response + .chunk() + .await + .map_err(|_| "Failed to read managed identity token response")? + { + if chunk.len() > MAX_TOKEN_RESPONSE_BYTES - body.len() { + return Err("Managed identity token response exceeds the size limit".into()); + } + body.extend_from_slice(&chunk); + } + CachedToken::parse(&body, identity, requested_at) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + use tokio::net::TcpListener; + + const CLIENT_ID: &str = "11111111-1111-1111-1111-111111111111"; + + #[test] + fn managed_identity_validates_client_id_configuration() { + assert_eq!(parse_client_id("").unwrap(), None); + let client_id = "12345678-1234-1234-1234-123456789abc"; + for value in [client_id.to_string(), client_id.to_ascii_uppercase()] { + assert_eq!( + parse_client_id(&value).unwrap().unwrap().to_string(), + client_id + ); + } + for invalid in [ + " ", + "\n", + "PRIVATE_VALUE", + "00000000-0000-0000-0000-000000000000", + "12345678-1234-1234-1234-123456789abc\n", + "12345678-1234-1234-1234-123456789abc,11111111-1111-1111-1111-111111111111", + ] { + let error = parse_client_id(invalid).unwrap_err(); + assert!(!error.contains("PRIVATE_VALUE")); + } + } + + fn http_response(body: &str) -> String { + format!("HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()) + } + + async fn mock_provider( + responses: Vec, + ) -> (TokenClient, tokio::task::JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let endpoint = format!( + "http://{}/metadata/identity/oauth2/token", + listener.local_addr().unwrap() + ); + let server = tokio::spawn(async move { + let mut requests = Vec::new(); + for response in responses { + let (stream, _) = tokio::time::timeout(Duration::from_secs(15), listener.accept()) + .await + .unwrap() + .unwrap(); + let mut stream = BufReader::new(stream); + let mut request = String::new(); + loop { + let mut line = String::new(); + let count = + tokio::time::timeout(Duration::from_secs(5), stream.read_line(&mut line)) + .await + .unwrap() + .unwrap(); + if count == 0 || line == "\r\n" { + break; + } + request.push_str(&line); + } + requests.push(request); + let _ = stream.get_mut().write_all(response.as_bytes()).await; + } + requests + }); + ( + TokenClient::new(&endpoint, parse_client_id(CLIENT_ID).unwrap()).unwrap(), + server, + ) + } + + fn identity() -> TokenResource { + TokenResource::for_endpoint(&Url::parse("https://account.blob.core.windows.net").unwrap()) + .unwrap() + } + + fn response() -> serde_json::Value { + serde_json::json!({ + "access_token": "PRIVATE_TOKEN", + "token_type": "Bearer", + "resource": "https://storage.azure.com/", + "expires_in": "3600" + }) + } + + #[tokio::test] + async fn managed_identity_cache_coalesces_concurrent_requests() { + let (client, server) = mock_provider(vec![http_response(&response().to_string())]).await; + let client = Arc::new(client); + assert!(client.client.get().is_none()); + let mut tasks = Vec::new(); + for _ in 0..16 { + let client = client.clone(); + tasks.push(tokio::spawn(async move { + client.authorization(&identity()).await.unwrap() + })); + } + for task in tasks { + let header = task.await.unwrap(); + assert_eq!(header, "Bearer PRIVATE_TOKEN"); + assert!(header.is_sensitive()); + } + let requests = server.await.unwrap(); + assert_eq!(requests.len(), 1); + assert!(requests[0] + .to_ascii_lowercase() + .contains("metadata: true\r\n")); + assert!(!requests[0].contains("PRIVATE_TOKEN")); + let url = Url::parse(&format!( + "http://localhost{}", + requests[0].split_whitespace().nth(1).unwrap() + )) + .unwrap(); + let query: BTreeMap<_, _> = url.query_pairs().into_owned().collect(); + assert_eq!(query.len(), 3); + assert_eq!(query["api-version"], "2018-02-01"); + assert_eq!(query["resource"], "https://storage.azure.com/"); + assert_eq!(query["client_id"], CLIENT_ID); + assert_eq!( + client.authorization(&identity()).await.unwrap(), + "Bearer PRIVATE_TOKEN" + ); + } + + #[tokio::test] + async fn managed_identity_cache_separates_clients_and_resources() { + let resources = [ + identity(), + TokenResource::for_endpoint(&Url::parse("https://example.openai.azure.com").unwrap()) + .unwrap(), + ]; + let responses = (0..2) + .flat_map(|client_index| { + resources + .iter() + .enumerate() + .map(move |(resource_index, resource)| { + let mut response = response(); + response["resource"] = serde_json::json!(resource.resource); + response["access_token"] = serde_json::json!(format!( + "PRIVATE_TOKEN_{client_index}_{resource_index}" + )); + http_response(&response.to_string()) + }) + }) + .collect(); + let (client, server) = mock_provider(responses).await; + let other_client = TokenClient::new( + client.endpoint.as_str(), + parse_client_id("22222222-2222-2222-2222-222222222222").unwrap(), + ) + .unwrap(); + let clients = [client, other_client]; + for _ in 0..2 { + for (client_index, client) in clients.iter().enumerate() { + for (resource_index, resource) in resources.iter().enumerate() { + assert_eq!( + client.authorization(resource).await.unwrap(), + format!("Bearer PRIVATE_TOKEN_{client_index}_{resource_index}") + ); + } + } + } + let requests = server.await.unwrap(); + assert_eq!(requests.len(), clients.len() * resources.len()); + for (request, (client, resource)) in requests.iter().zip( + clients + .iter() + .flat_map(|client| resources.iter().map(move |resource| (client, resource))), + ) { + let url = Url::parse(&format!( + "http://localhost{}", + request.split_whitespace().nth(1).unwrap() + )) + .unwrap(); + let query: BTreeMap<_, _> = url.query_pairs().into_owned().collect(); + assert_eq!(query["resource"], resource.resource); + assert_eq!(query["client_id"], client.client_id.unwrap().to_string()); + } + } + + #[tokio::test] + async fn managed_identity_cache_refreshes_and_never_serves_stale_on_failure() { + let responses = ["INITIAL", "REFRESHED", "REFRESHED_AGAIN"].map(|token| { + let mut response = response(); + response["access_token"] = serde_json::json!(token); + http_response(&response.to_string()) + }); + let (client, server) = mock_provider(responses.into_iter().chain([ + "HTTP/1.1 503 Unavailable\r\nContent-Length: 13\r\nConnection: close\r\n\r\nPRIVATE_TOKEN".into(), + ]).collect()).await; + assert_eq!( + client.authorization(&identity()).await.unwrap(), + "Bearer INITIAL" + ); + let entry = client.tokens.lock().await.get(&identity()).unwrap().clone(); + entry.lock().await.as_mut().unwrap().refresh_at = Instant::now(); + assert_eq!( + client.authorization(&identity()).await.unwrap(), + "Bearer REFRESHED" + ); + entry.lock().await.as_mut().unwrap().refresh_by = UNIX_EPOCH; + assert_eq!( + client.authorization(&identity()).await.unwrap(), + "Bearer REFRESHED_AGAIN" + ); + entry.lock().await.as_mut().unwrap().refresh_at = Instant::now(); + let error = client.authorization(&identity()).await.unwrap_err(); + assert!(error.contains("HTTP 503")); + assert!(!error.contains("PRIVATE_TOKEN")); + assert!(entry.lock().await.is_none()); + assert_eq!(server.await.unwrap().len(), 4); + } + + #[tokio::test] + async fn managed_identity_provider_failures_are_bounded_and_redacted() { + let oversized = "x".repeat(MAX_TOKEN_RESPONSE_BYTES + 1); + let responses = vec![ + ("HTTP/1.1 302 Found\r\nLocation: http://127.0.0.1:9/?token=PRIVATE_TOKEN\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".into(), "HTTP 302"), + ("HTTP/1.1 400 Bad Request\r\nContent-Length: 13\r\nConnection: close\r\n\r\nPRIVATE_TOKEN".into(), "HTTP 400"), + (http_response("{\"error\":\"PRIVATE_TOKEN\",\"error_description\":\"PRIVATE_TOKEN\"}"), "Invalid managed identity token response"), + (http_response("PRIVATE_TOKEN"), "Invalid managed identity token response"), + (format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", MAX_TOKEN_RESPONSE_BYTES + 1), "size limit"), + (format!("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n{:x}\r\n{oversized}\r\n0\r\n\r\n", oversized.len()), "size limit"), + ("HTTP/1.1 200 OK\r\nContent-Length: 100\r\nConnection: close\r\n\r\nPRIVATE_TOKEN".into(), "Failed to read managed identity token response"), + ]; + let (client, server) = mock_provider( + responses + .iter() + .map(|(response, _)| response.clone()) + .collect(), + ) + .await; + for (_, expected) in responses { + let error = client.authorization(&identity()).await.unwrap_err(); + assert!(error.contains(expected), "{error}"); + assert!(!error.contains("PRIVATE_TOKEN")); + } + assert_eq!(server.await.unwrap().len(), 7); + } + + #[tokio::test] + async fn managed_identity_unset_disables_token_acquisition() { + let client = TokenClient::new("http://127.0.0.1:9/token", None).unwrap(); + let mut request = crate::endpoints::EndpointRequest { + url: Url::parse("https://account.blob.core.windows.net/").unwrap(), + credential_header: None, + managed_identity: Some(identity()), + }; + let error = request.authorize(&client).await.unwrap_err(); + assert!(error.contains("configure pg_durable.managed_identity_client_id")); + assert!(request.credential_header.is_none()); + assert!(client.client.get().is_none()); + assert!(client.tokens.lock().await.is_empty()); + } + + #[tokio::test] + async fn managed_identity_timeout_releases_waiters() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let client = TokenClient::new( + &format!("http://{}/token", listener.local_addr().unwrap()), + parse_client_id(CLIENT_ID).unwrap(), + ) + .unwrap(); + let server = tokio::spawn(async move { + let (first, _) = listener.accept().await.unwrap(); + let (mut second, _) = listener.accept().await.unwrap(); + second + .write_all(http_response(&response().to_string()).as_bytes()) + .await + .unwrap(); + drop(first); + }); + let error = client.authorization(&identity()).await.unwrap_err(); + assert!(error.contains("timed out"), "{error}"); + assert_eq!( + client.authorization(&identity()).await.unwrap(), + "Bearer PRIVATE_TOKEN" + ); + server.await.unwrap(); + } + + #[tokio::test] + async fn managed_identity_is_not_contacted_for_other_authentication() { + let client = TokenClient::new(DEFAULT_TOKEN_ENDPOINT.to_str().unwrap(), None).unwrap(); + let mut request = crate::endpoints::EndpointRequest { + url: Url::parse("https://api.github.com/").unwrap(), + credential_header: None, + managed_identity: None, + }; + request.authorize(&client).await.unwrap(); + assert!(client.client.get().is_none()); + assert!(client.tokens.lock().await.is_empty()); + } + + #[test] + fn managed_identity_derives_resource_from_destination() { + for (url, resource) in [ + ( + "https://example.openai.azure.com", + "https://cognitiveservices.azure.com", + ), + ( + "https://example.cognitiveservices.azure.com", + "https://cognitiveservices.azure.com", + ), + ( + "https://example.services.ai.azure.com", + "https://cognitiveservices.azure.com", + ), + ( + "https://account.blob.core.windows.net", + "https://storage.azure.com/", + ), + ( + "https://account.z01.blob.storage.azure.net", + "https://storage.azure.com/", + ), + ( + "https://account.dfs.core.windows.net", + "https://storage.azure.com/", + ), + ( + "https://account.queue.core.windows.net", + "https://storage.azure.com/", + ), + ( + "https://account.table.core.windows.net", + "https://storage.azure.com/", + ), + ( + "https://account.file.core.windows.net", + "https://storage.azure.com/", + ), + ("https://example.vault.azure.net", "https://vault.azure.net"), + ( + "https://management.azure.com", + "https://management.azure.com/", + ), + ] { + assert_eq!( + TokenResource::for_endpoint(&Url::parse(url).unwrap()) + .unwrap() + .resource, + resource + ); + } + for url in [ + "http://account.blob.core.windows.net", + "https://account.blob.core.windows.net:8443", + "https://account.blob.core.windows.net.evil.test", + "https://notblob.core.windows.net", + "https://blob.core.windows.net", + "https://account.blob.core.windows.net.", + "https://api.github.com", + "https://example.azurewebsites.net", + "https://169.254.169.254", + ] { + assert!( + TokenResource::for_endpoint(&Url::parse(url).unwrap()).is_err(), + "{url}" + ); + } + } + + #[test] + fn managed_identity_validates_provider_configuration() { + for valid in [ + DEFAULT_TOKEN_ENDPOINT.to_str().unwrap(), + "http://127.0.0.1:8080/token", + "http://[::1]:8080/token", + "https://identity.example/token", + ] { + validate_token_endpoint(valid).unwrap(); + } + for invalid in [ + "http://identity.example/token", + "file:///token", + "https://user:PRIVATE_VALUE@identity.example", + "https://identity.example/?key=PRIVATE_VALUE", + "https://identity.example/#PRIVATE_VALUE", + "https://identity.example/\n", + ] { + let error = validate_token_endpoint(invalid).unwrap_err(); + assert!(!error.contains("PRIVATE_VALUE")); + } + } + + #[test] + fn managed_identity_token_is_sensitive_and_expiring() { + let token = CachedToken::parse( + response().to_string().as_bytes(), + &identity(), + SystemTime::now(), + ) + .unwrap(); + assert_eq!(token.authorization, "Bearer PRIVATE_TOKEN"); + assert!(token.authorization.is_sensitive()); + assert!(!format!("{:?}", token.authorization).contains("PRIVATE_TOKEN")); + assert!(token.is_fresh()); + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs(); + for expiry in [ + serde_json::json!(now + 3600), + serde_json::json!((now + 3600).to_string()), + ] { + let mut response = response(); + response["expires_on"] = expiry; + assert!(CachedToken::parse( + response.to_string().as_bytes(), + &identity(), + SystemTime::now() + ) + .is_ok()); + } + } + + #[test] + fn managed_identity_rejects_invalid_tokens_without_echoing_them() { + for (field, value) in [ + ( + "access_token", + serde_json::json!("PRIVATE_TOKEN\r\nInjected: value"), + ), + ("access_token", serde_json::json!("")), + ("token_type", serde_json::json!("PRIVATE_TOKEN")), + ("resource", serde_json::json!("https://vault.azure.net")), + ("expires_on", serde_json::json!("PRIVATE_TOKEN")), + ("expires_on", serde_json::json!(0)), + ("expires_on", serde_json::json!(-1)), + ("expires_in", serde_json::json!(30)), + ("expires_in", serde_json::Value::Null), + ("not_before", serde_json::json!(u64::MAX)), + ("error", serde_json::json!("PRIVATE_TOKEN")), + ] { + let mut response = response(); + response[field] = value; + let error = CachedToken::parse( + response.to_string().as_bytes(), + &identity(), + SystemTime::now(), + ) + .err() + .expect("invalid response must fail"); + assert!(!error.contains("PRIVATE_TOKEN"), "{error}"); + } + assert!(CachedToken::parse(b"PRIVATE_TOKEN", &identity(), SystemTime::now()).is_err()); + } +} diff --git a/src/registry.rs b/src/registry.rs index 69ca80eb..f16e0feb 100644 --- a/src/registry.rs +++ b/src/registry.rs @@ -18,6 +18,7 @@ pub fn create_activity_registry( pool: Arc, semaphore: Arc, http_policy: Arc, + identity_client: Arc, ) -> ActivityRegistry { let sql_semaphore = semaphore.clone(); let http_semaphore = semaphore.clone(); @@ -31,6 +32,7 @@ pub fn create_activity_registry( let node_status_pool = router.clone(); let http_pool = router.clone(); let multipart_pool = router; + let multipart_identity_client = identity_client.clone(); ActivityRegistry::builder() .register( @@ -149,6 +151,7 @@ pub fn create_activity_registry( let pool = http_pool.clone(); let semaphore = http_semaphore.clone(); let policy = http_policy.clone(); + let identity_client = identity_client.clone(); async move { let mut route = pool.route(ctx.instance_id()).await?; let result = activities::execute_http::execute( @@ -156,6 +159,7 @@ pub fn create_activity_registry( &mut route, semaphore, policy, + identity_client, config_json, ) .await; @@ -170,6 +174,7 @@ pub fn create_activity_registry( let pool = multipart_pool.clone(); let semaphore = multipart_semaphore.clone(); let policy = multipart_policy.clone(); + let identity_client = multipart_identity_client.clone(); async move { let mut route = pool.route(ctx.instance_id()).await?; let result = activities::execute_multipart::execute( @@ -177,6 +182,7 @@ pub fn create_activity_registry( &mut route, semaphore, policy, + identity_client, config_json, ) .await; diff --git a/src/secrets.rs b/src/secrets.rs index a4a2f51e..8f156ccd 100644 --- a/src/secrets.rs +++ b/src/secrets.rs @@ -213,7 +213,7 @@ impl SecretOptions { ); } } - if let Some((name, _)) = &request.credential_header { + if let Some(name) = request.credential_header_name() { if bindings .headers .keys() @@ -397,6 +397,7 @@ mod unit_tests { let mut request = crate::endpoints::EndpointRequest { url: url::Url::parse("https://api.github.com/?sig=existing%2Bvalue").unwrap(), credential_header: None, + managed_identity: None, }; options.validate(false, "POST", false, None).unwrap(); options.validate_destinations(&request, None).unwrap(); @@ -435,6 +436,7 @@ mod unit_tests { let mut request = crate::endpoints::EndpointRequest { url: url::Url::parse("https://api.github.com/?%6bey=ordinary").unwrap(), credential_header: None, + managed_identity: None, }; assert!(options.validate_destinations(&request, None).is_err()); request.url.set_query(None); @@ -449,6 +451,20 @@ mod unit_tests { reqwest::header::HeaderValue::from_static("endpoint"), )); assert!(options.validate_destinations(&request, None).is_err()); + request.credential_header = None; + request.managed_identity = Some( + crate::managed_identity::TokenResource::for_endpoint( + &url::Url::parse("https://account.blob.core.windows.net").unwrap(), + ) + .unwrap(), + ); + let identity_options: SecretOptions = serde_json::from_value( + json!({"secret_bindings":{"headers":{"aUtHoRiZaTiOn":{"server":"foo","key":"bar"}}}}), + ) + .unwrap(); + assert!(identity_options + .validate_destinations(&request, None) + .is_err()); let form: SecretOptions = serde_json::from_value(json!({"form_fields":{}})).unwrap(); assert!(form.validate(true, "POST", false, None).is_err()); assert!(form.validate(false, "POST", true, None).is_err()); @@ -480,6 +496,7 @@ mod unit_tests { let mut request = crate::endpoints::EndpointRequest { url: url::Url::parse("https://api.github.com/").unwrap(), credential_header: None, + managed_identity: None, }; let catalog = BTreeMap::from([( "foo".into(), diff --git a/src/worker.rs b/src/worker.rs index 19d372cf..21a809d6 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -112,6 +112,29 @@ pub extern "C-unwind" fn duroxide_worker_main(_arg: pg_sys::Datum) { |err| error!("invalid value for parameter \"pg_durable.http_allowed_domains\": {err}"), ), }); + let configured_identity_endpoint = crate::MANAGED_IDENTITY_ENDPOINT + .get() + .unwrap_or_else(|| error!("pg_durable.managed_identity_endpoint must not be NULL")); + let configured_client_id = crate::MANAGED_IDENTITY_CLIENT_ID.get().unwrap_or_default(); + let client_id = crate::managed_identity::parse_client_id( + configured_client_id + .to_str() + .unwrap_or_else(|_| error!("pg_durable.managed_identity_client_id must be UTF-8")), + ) + .unwrap_or_else(|err| { + error!("invalid value for parameter \"pg_durable.managed_identity_client_id\": {err}") + }); + let identity_client = Arc::new( + crate::managed_identity::TokenClient::new( + configured_identity_endpoint + .to_str() + .unwrap_or_else(|_| error!("pg_durable.managed_identity_endpoint must be UTF-8")), + client_id, + ) + .unwrap_or_else(|err| { + error!("invalid value for parameter \"pg_durable.managed_identity_endpoint\": {err}") + }), + ); let rt = match tokio::runtime::Builder::new_current_thread() .enable_all() @@ -125,7 +148,7 @@ pub extern "C-unwind" fn duroxide_worker_main(_arg: pg_sys::Datum) { }; rt.block_on(async { - run_duroxide_runtime(http_policy).await; + run_duroxide_runtime(http_policy, identity_client).await; }); // All async cleanup (pool closes, runtime shutdown) is performed inside @@ -139,7 +162,10 @@ pub extern "C-unwind" fn duroxide_worker_main(_arg: pg_sys::Datum) { } /// Run the duroxide runtime with proper shutdown handling -async fn run_duroxide_runtime(http_policy: Arc) { +async fn run_duroxide_runtime( + http_policy: Arc, + identity_client: Arc, +) { const WAIT_FOR_EXTENSION_POLL_INTERVAL: Duration = Duration::from_secs(5); const EXTENSION_DROP_POLL_INTERVAL: Duration = Duration::from_secs(5); const INIT_RETRY_INTERVAL: Duration = Duration::from_secs(1); @@ -304,6 +330,7 @@ async fn run_duroxide_runtime(http_policy: Arc) { &mgmt_pool, &duroxide_schema, &http_policy, + &identity_client, ) .await else { @@ -693,6 +720,7 @@ async fn initialize_duroxide_runtime( mgmt_pool: &sqlx::PgPool, schema_name: &str, http_policy: &Arc, + identity_client: &Arc, ) -> Option<(Arc, Arc)> { log!("pg_durable: initializing duroxide runtime..."); @@ -785,6 +813,7 @@ async fn initialize_duroxide_runtime( Arc::new(mgmt_pool.clone()), user_semaphore.clone(), http_policy.clone(), + identity_client.clone(), ); let orchestrations = create_orchestration_registry(); diff --git a/tests/e2e/managed_identity_mock.py b/tests/e2e/managed_identity_mock.py new file mode 100644 index 00000000..8d69af10 --- /dev/null +++ b/tests/e2e/managed_identity_mock.py @@ -0,0 +1,185 @@ +import argparse +import http.client +import http.server +import json +from pathlib import Path +import socketserver +import ssl +import subprocess +import sys +import threading +import time +from urllib.parse import parse_qs, urlsplit + + +HOST = "pg-durable-mi.blob.core.windows.net" +CLIENT_ID = "11111111-1111-1111-1111-111111111111" +RESOURCE = "https://storage.azure.com/" +FAILING_RESOURCE = "https://vault.azure.net" +TOKEN = "MI_PRIVATE_USER_TOKEN" + + +class State: + def __init__(self): + self.lock = threading.Lock() + self.token_requests = {} + self.requests = [] + + +class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *_args): + pass + + def reply(self, status, body=None, headers=None): + encoded = json.dumps(body).encode() if body is not None else b"" + self.send_response(status) + self.send_header("Content-Length", str(len(encoded))) + self.send_header("Content-Type", "application/json") + self.send_header("Connection", "close") + for name, value in (headers or {}).items(): + self.send_header(name, value) + self.end_headers() + self.wfile.write(encoded) + self.close_connection = True + + +class TokenHandler(Handler): + def do_GET(self): + url = urlsplit(self.path) + query = parse_qs(url.query, keep_blank_values=True) + if ( + url.path != "/metadata/identity/oauth2/token" + or self.headers.get("Metadata") != "true" + or query.get("api-version") != ["2018-02-01"] + or query.get("client_id") != [CLIENT_ID] + or query.get("resource") not in ([RESOURCE], [FAILING_RESOURCE]) + or set(query) - {"resource", "api-version", "client_id"} + ): + self.reply(400, {"error": "invalid_request"}) + return + resource = query["resource"][0] + with self.server.state.lock: + counts = self.server.state.token_requests + counts[resource] = counts.get(resource, 0) + 1 + if resource == FAILING_RESOURCE: + self.reply(400, {"error": "invalid_request", "error_description": "MI_PRIVATE_PROVIDER_ERROR"}) + return + self.reply(200, { + "access_token": TOKEN, + "token_type": "Bearer", + "resource": RESOURCE, + "expires_on": str(int(time.time()) + 3600), + }) + + +class DestinationHandler(Handler): + def do_GET(self): + self.respond() + + def do_POST(self): + self.respond() + + def respond(self): + path = urlsplit(self.path).path + if self.headers.get_all("Authorization") != ["Bearer " + TOKEN]: + self.reply(401, {"error": "unexpected authorization"}) + return + length = int(self.headers.get("Content-Length", "0")) + if length > 8192: + self.reply(413) + return + body = self.rfile.read(length) + multipart = self.headers.get("Content-Type", "").startswith("multipart/form-data;") + if self.command == "POST" and (not multipart or b"hello" not in body): + self.reply(400, {"error": "expected multipart upload"}) + return + with self.server.state.lock: + if path == "/system/stats": + result = { + "token_requests": dict(self.server.state.token_requests), + "requests": list(self.server.state.requests), + } + else: + self.server.state.requests.append({"method": self.command, "path": path, "multipart": multipart}) + result = None + if path == "/system/stats": + self.reply(200, result) + elif path.endswith("/redirect"): + self.reply(302, headers={"Location": "https://unapproved.invalid/"}) + elif path.endswith("/unauthorized"): + self.reply(401) + else: + self.reply(204) + + +class ProxyHandler(socketserver.StreamRequestHandler): + def handle(self): + self.request.settimeout(15) + request = self.rfile.readline(4096).decode("ascii").strip() + while self.rfile.readline(4096) not in (b"\r\n", b"", b"\n"): + pass + if request != f"CONNECT {HOST}:443 HTTP/1.1": + self.wfile.write(b"HTTP/1.1 403 Forbidden\r\nContent-Length: 0\r\n\r\n") + return + self.wfile.write(b"HTTP/1.1 200 Connection Established\r\n\r\n") + self.wfile.flush() + with self.server.tls.wrap_socket(self.request, server_side=True) as connection: + DestinationHandler(connection, self.client_address, self.server) + + +class ProxyServer(socketserver.ThreadingTCPServer): + daemon_threads = True + + +def self_test(token_server, proxy_server, certificate): + connection = http.client.HTTPConnection(*token_server.server_address, timeout=5) + connection.request("GET", f"/metadata/identity/oauth2/token?api-version=2018-02-01&resource=https%3A%2F%2Fstorage.azure.com%2F&client_id={CLIENT_ID}", headers={"Metadata": "true"}) + response = connection.getresponse() + assert response.status == 200 + assert json.loads(response.read())["access_token"] == TOKEN + connection.close() + context = ssl.create_default_context(cafile=certificate) + connection = http.client.HTTPSConnection(*proxy_server.server_address, context=context, timeout=5) + connection.set_tunnel(HOST, 443) + connection.request("GET", "/system/data", headers={"Authorization": "Bearer " + TOKEN}) + response = connection.getresponse() + assert response.status == 204 + response.read() + connection.close() + print("Managed identity mock self-test passed", flush=True) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("directory", type=Path) + parser.add_argument("--self-test", action="store_true") + args = parser.parse_args() + args.directory.mkdir(parents=True, exist_ok=True) + certificate = str(args.directory / "certificate.pem") + private_key = str(args.directory / "key.pem") + subprocess.run([ + "openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes", "-days", "1", + "-subj", f"/CN={HOST}", "-addext", f"subjectAltName=DNS:{HOST}", + "-keyout", private_key, "-out", certificate, + ], check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) + state = State() + with http.server.ThreadingHTTPServer(("127.0.0.1", 0), TokenHandler) as token_server, ProxyServer(("127.0.0.1", 0), ProxyHandler) as proxy_server: + token_server.state = state + proxy_server.state = state + proxy_server.tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + proxy_server.tls.load_cert_chain(certificate, private_key) + for server in (token_server, proxy_server): + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + if args.self_test: + self_test(token_server, proxy_server, certificate) + else: + print(f"http://127.0.0.1:{token_server.server_port}/metadata/identity/oauth2/token\thttp://127.0.0.1:{proxy_server.server_address[1]}\t{certificate}", flush=True) + sys.stdin.readline() + finally: + token_server.shutdown() + proxy_server.shutdown() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/e2e/sql/47_http_dsl_disabled.sql b/tests/e2e/sql/47_http_dsl_disabled.sql index af533011..8f4ced77 100644 --- a/tests/e2e/sql/47_http_dsl_disabled.sql +++ b/tests/e2e/sql/47_http_dsl_disabled.sql @@ -138,6 +138,38 @@ BEGIN END $$; DROP TABLE _test_http_bypass; + +RESET SESSION AUTHORIZATION; +CREATE SERVER mi_http_disabled FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://pg-durable-mi.blob.core.windows.net', auth_scheme 'managed-identity'); +GRANT USAGE ON FOREIGN SERVER mi_http_disabled TO df_e2e_user; +SET SESSION AUTHORIZATION df_e2e_user; +CREATE TEMP TABLE _test_mi_disabled AS +SELECT df.start(jsonb_build_object( + 'node_type', node_type, + 'query', jsonb_build_object('endpoint', 'mi_http_disabled', 'url', '/data', 'method', 'POST', + 'parts', '[{"name":"file","data_b64":"aGVsbG8="}]'::jsonb)::text +)::text, 'mi-disabled-' || node_type) AS instance_id +FROM (VALUES ('HTTP'), ('HTTP_MULTIPART')) AS node_types(node_type); + +DO $$ +DECLARE + test_case record; + actual_status text; + node_result text; +BEGIN + FOR test_case IN SELECT * FROM _test_mi_disabled LOOP + actual_status := df.await_instance(test_case.instance_id, 30); + SELECT result::text INTO node_result FROM df.nodes + WHERE instance_id = test_case.instance_id AND node_type IN ('HTTP', 'HTTP_MULTIPART'); + IF actual_status IS DISTINCT FROM 'failed' OR node_result IS NULL + OR node_result NOT ILIKE '%outbound HTTP requests are disabled%' THEN + RAISE EXCEPTION 'TEST FAILED: managed identity bypassed the HTTP kill switch: %, %', actual_status, node_result; + END IF; + END LOOP; +END $$; +DROP TABLE _test_mi_disabled; RESET SESSION AUTHORIZATION; +DROP SERVER mi_http_disabled; SELECT 'TEST PASSED' AS result; diff --git a/tests/e2e/sql/69_http_allowed_domains.sql b/tests/e2e/sql/69_http_allowed_domains.sql index 11479056..170f2fec 100644 --- a/tests/e2e/sql/69_http_allowed_domains.sql +++ b/tests/e2e/sql/69_http_allowed_domains.sql @@ -159,6 +159,10 @@ BEGIN END $$; DROP TABLE _test_http_security_reload; +CREATE SERVER mi_custom_domains FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://pg-durable-mi.blob.core.windows.net', auth_scheme 'managed-identity'); +GRANT USAGE ON FOREIGN SERVER mi_custom_domains TO df_e2e_user; + SET SESSION AUTHORIZATION df_e2e_user; CREATE TEMP TABLE _test_http_allowed_domains ( @@ -187,6 +191,17 @@ FROM (VALUES ) AS endpoints(hostname, allowed) CROSS JOIN (VALUES ('HTTP'), ('HTTP_MULTIPART')) AS node_types(node_type); +INSERT INTO _test_http_allowed_domains +SELECT df.start( + CASE node_type + WHEN 'HTTP' THEN df.http(df.endpoint('mi_custom_domains', '/data'), 'GET') + ELSE df.http_multipart(df.endpoint('mi_custom_domains', '/upload'), + parts => '[{"name":"file","data_b64":"aGVsbG8="}]') + END, + 'managed-identity-custom-domains-' || node_type +), 'managed-identity', node_type, false +FROM (VALUES ('HTTP'), ('HTTP_MULTIPART')) AS node_types(node_type); + DO $$ DECLARE test_case RECORD; @@ -236,5 +251,6 @@ END $$; DROP TABLE _test_http_allowed_domains; RESET SESSION AUTHORIZATION; +DROP SERVER mi_custom_domains; SELECT 'TEST PASSED' AS result; diff --git a/tests/e2e/sql/72_endpoint_catalog.sql b/tests/e2e/sql/72_endpoint_catalog.sql index 3375ff6b..6ec62acb 100644 --- a/tests/e2e/sql/72_endpoint_catalog.sql +++ b/tests/e2e/sql/72_endpoint_catalog.sql @@ -36,10 +36,20 @@ BEGIN BEGIN ALTER SERVER ec_server OPTIONS (SET auth_scheme 'managed-identity', DROP header_name); EXCEPTION WHEN OTHERS THEN - IF SQLERRM NOT LIKE '%not supported in this version%' THEN RAISE; END IF; + IF SQLERRM NOT LIKE '%not supported for this destination hostname%' THEN RAISE; END IF; rejected := true; END; - IF NOT rejected THEN RAISE EXCEPTION 'TEST FAILED: managed identity enabled without controls'; END IF; + IF NOT rejected THEN RAISE EXCEPTION 'TEST FAILED: managed identity accepted an unmapped destination'; END IF; + + rejected := false; + BEGIN + ALTER SERVER ec_server OPTIONS (SET base_url 'https://account.blob.core.windows.net', + SET auth_scheme 'managed-identity', DROP header_name); + EXCEPTION WHEN OTHERS THEN + IF SQLERRM NOT LIKE '%EXECUTE on df.managed_identity_admin()%' THEN RAISE; END IF; + rejected := true; + END; + IF NOT rejected THEN RAISE EXCEPTION 'TEST FAILED: delegated endpoint owner enabled managed identity'; END IF; rejected := false; BEGIN diff --git a/tests/e2e/sql/75_managed_identity.sql b/tests/e2e/sql/75_managed_identity.sql new file mode 100644 index 00000000..d26172f4 --- /dev/null +++ b/tests/e2e/sql/75_managed_identity.sql @@ -0,0 +1,269 @@ +RESET SESSION AUTHORIZATION; +DROP SERVER IF EXISTS mi_system, mi_user, mi_error, mi_denied CASCADE; +DO $$ BEGIN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'mi_no_http') THEN + DROP OWNED BY mi_no_http; + END IF; + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'mi_endpoint_admin') THEN + DROP OWNED BY mi_endpoint_admin; + END IF; +END $$; +DROP ROLE IF EXISTS mi_no_http, mi_endpoint_admin; +CREATE ROLE mi_no_http LOGIN; +CREATE ROLE mi_endpoint_admin LOGIN; +SELECT df.grant_usage('mi_no_http'); +SELECT df.grant_usage('mi_endpoint_admin', include_http => true, with_grant => true); +SELECT df.grant_usage('df_e2e_user', include_http => true); + +DO $$ +DECLARE + statement text; +BEGIN + IF NOT EXISTS (SELECT 1 FROM pg_settings WHERE name = 'pg_durable.managed_identity_endpoint' + AND context = 'postmaster' AND source = 'configuration file' AND NOT pending_restart) THEN + RAISE EXCEPTION 'TEST FAILED: managed identity provider must be a startup-only setting'; + END IF; + IF NOT EXISTS (SELECT 1 FROM pg_settings WHERE name = 'pg_durable.managed_identity_client_id' + AND setting = '11111111-1111-1111-1111-111111111111' AND boot_val = '' + AND context = 'postmaster' AND source = 'configuration file' AND NOT pending_restart) THEN + RAISE EXCEPTION 'TEST FAILED: managed identity must be pinned at startup and default to disabled'; + END IF; + IF pg_catalog.has_function_privilege('mi_endpoint_admin', 'df.managed_identity_admin()', 'EXECUTE') + OR pg_catalog.has_foreign_data_wrapper_privilege('mi_endpoint_admin', 'pg_durable_fdw', 'USAGE') THEN + RAISE EXCEPTION 'TEST FAILED: general administration grants enabled managed identity configuration'; + END IF; + FOREACH statement IN ARRAY ARRAY[ + 'SET pg_durable.managed_identity_endpoint = ''http://127.0.0.1:9/token''', + 'ALTER ROLE df_e2e_user SET pg_durable.managed_identity_endpoint = ''http://127.0.0.1:9/token''', + format('ALTER DATABASE %I SET pg_durable.managed_identity_endpoint = ''http://127.0.0.1:9/token''', current_database()), + 'SET pg_durable.managed_identity_client_id = ''22222222-2222-2222-2222-222222222222''', + 'SET LOCAL pg_durable.managed_identity_client_id = ''22222222-2222-2222-2222-222222222222''', + 'ALTER ROLE df_e2e_user SET pg_durable.managed_identity_client_id = ''22222222-2222-2222-2222-222222222222''', + format('ALTER DATABASE %I SET pg_durable.managed_identity_client_id = ''22222222-2222-2222-2222-222222222222''', current_database()) + ] LOOP + BEGIN + EXECUTE statement; + RAISE EXCEPTION 'TEST FAILED: accepted a runtime managed identity provider override'; + EXCEPTION WHEN cant_change_runtime_param THEN + NULL; + END; + END LOOP; +END $$; + +ALTER SYSTEM SET pg_durable.managed_identity_client_id = '22222222-2222-2222-2222-222222222222'; +SELECT pg_reload_conf(); +CREATE TEMP TABLE _mi_reload (unchanged boolean); +DO $$ +DECLARE + attempts integer := 0; +BEGIN + LOOP + EXIT WHEN EXISTS (SELECT 1 FROM pg_settings + WHERE name = 'pg_durable.managed_identity_client_id' AND pending_restart) + OR attempts >= 100; + PERFORM pg_sleep(0.05); + attempts := attempts + 1; + END LOOP; + INSERT INTO _mi_reload SELECT setting = '11111111-1111-1111-1111-111111111111' AND pending_restart + FROM pg_settings WHERE name = 'pg_durable.managed_identity_client_id'; +END $$; +ALTER SYSTEM RESET pg_durable.managed_identity_client_id; +SELECT pg_reload_conf(); +DO $$ +BEGIN + IF NOT EXISTS (SELECT 1 FROM _mi_reload WHERE unchanged) THEN + RAISE EXCEPTION 'TEST FAILED: client ID changed without a server restart'; + END IF; +END $$; +DROP TABLE _mi_reload; + +CREATE SERVER mi_system FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://pg-durable-mi.blob.core.windows.net/system', auth_scheme 'managed-identity'); +CREATE SERVER mi_error FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://pg-durable-mi.vault.azure.net/error', auth_scheme 'managed-identity'); +CREATE SERVER mi_denied FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://pg-durable-mi.blob.core.windows.net/system', auth_scheme 'managed-identity'); +GRANT USAGE ON FOREIGN SERVER mi_system, mi_error TO df_e2e_user; +GRANT USAGE ON FOREIGN SERVER mi_system TO mi_no_http; + +GRANT USAGE ON FOREIGN DATA WRAPPER pg_durable_fdw TO mi_endpoint_admin; +GRANT EXECUTE ON FUNCTION df.managed_identity_admin() TO mi_endpoint_admin; +SET SESSION AUTHORIZATION mi_endpoint_admin; +SELECT df.managed_identity_admin(); +CREATE SERVER mi_user FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://pg-durable-mi.blob.core.windows.net/user', auth_scheme 'managed-identity'); +GRANT USAGE ON FOREIGN SERVER mi_user TO df_e2e_user; +DO $$ +BEGIN + IF EXISTS (SELECT 1 FROM pg_settings WHERE name IN ('pg_durable.managed_identity_client_id', 'pg_durable.managed_identity_endpoint')) THEN + RAISE EXCEPTION 'TEST FAILED: endpoint administration exposed protected provider settings'; + END IF; + BEGIN + SET pg_durable.managed_identity_client_id = '22222222-2222-2222-2222-222222222222'; + RAISE EXCEPTION 'TEST FAILED: endpoint administrator changed the configured identity'; + EXCEPTION WHEN insufficient_privilege OR cant_change_runtime_param THEN + NULL; + END; + BEGIN + ALTER SERVER mi_user OPTIONS (ADD client_id '22222222-2222-2222-2222-222222222222'); + RAISE EXCEPTION 'TEST FAILED: endpoint administrator selected an identity'; + EXCEPTION WHEN OTHERS THEN + IF SQLERRM NOT LIKE '%Unsupported endpoint server option%' THEN RAISE; END IF; + END; +END $$; +RESET SESSION AUTHORIZATION; + +CREATE TEMP TABLE _mi_cases (instance_id text, expected text, status_code integer, error_pattern text); +GRANT SELECT, INSERT ON _mi_cases TO df_e2e_user, mi_no_http; +SET SESSION AUTHORIZATION df_e2e_user; + +INSERT INTO _mi_cases VALUES + (df.start(df.http(df.endpoint('mi_system', '/data'), 'GET'), 'mi-system-http'), 'completed', 204, NULL), + (df.start(df.http_multipart(df.endpoint('mi_system', '/upload'), parts => '[{"name":"file","data_b64":"aGVsbG8="}]'), 'mi-system-multipart'), 'completed', 204, NULL), + (df.start(df.http(df.endpoint('mi_user', '/data'), 'GET'), 'mi-user-http'), 'completed', 204, NULL), + (df.start(df.http_multipart(df.endpoint('mi_user', '/upload'), parts => '[{"name":"file","data_b64":"aGVsbG8="}]'), 'mi-user-multipart'), 'completed', 204, NULL), + (df.start(df.http(df.endpoint('mi_system', '/redirect'), 'GET'), 'mi-no-redirect'), 'completed', 302, NULL), + (df.start(df.http(df.endpoint('mi_system', '/unauthorized'), 'GET'), 'mi-unauthorized'), 'completed', 401, NULL), + (df.start(df.http(df.endpoint('mi_error', '/data'), 'GET'), 'mi-provider-error'), 'failed', NULL, '%token provider returned HTTP 400%'), + (df.start(df.http_multipart(df.endpoint('mi_error', '/upload'), parts => '[{"name":"file","data_b64":"aGVsbG8="}]'), 'mi-provider-error-multipart'), 'failed', NULL, '%token provider returned HTTP 400%'), + (df.start(df.http(df.endpoint('mi_denied', '/data'), 'GET'), 'mi-usage-denied'), 'failed', NULL, '%USAGE%'), + (df.start(df.http_multipart(df.endpoint('mi_denied', '/upload'), parts => '[{"name":"file","data_b64":"aGVsbG8="}]'), 'mi-usage-denied-multipart'), 'failed', NULL, '%USAGE%'), + (df.start(df.http(df.endpoint('mi_system', '/data'), 'GET', headers => '{"aUtHoRiZaTiOn":"override"}'), 'mi-override'), 'failed', NULL, '%override endpoint%'), + (df.start(df.http_multipart(df.endpoint('mi_system', '/upload'), parts => '[{"name":"file","data_b64":"aGVsbG8="}]', headers => '{"Authorization":"override"}'), 'mi-override-multipart'), 'failed', NULL, '%override endpoint%'), + (df.start(df.with_http_options(df.http(df.endpoint('mi_system', '/data'), 'GET'), + jsonb_build_object('secret_bindings', jsonb_build_object('headers', jsonb_build_object('Authorization', df.secret('mi_system', 'absent')))))), 'failed', NULL, '%override endpoint authentication%'), + (df.start(df.with_http_options(df.http_multipart(df.endpoint('mi_system', '/upload'), parts => '[{"name":"file","data_b64":"aGVsbG8="}]'), + jsonb_build_object('secret_bindings', jsonb_build_object('headers', jsonb_build_object('authorization', df.secret('mi_system', 'absent')))))), 'failed', NULL, '%override endpoint authentication%'); + +INSERT INTO _mi_cases +SELECT df.start(jsonb_build_object('node_type', node_type, 'query', jsonb_build_object( + 'endpoint', 'mi_user', 'url', CASE WHEN node_type = 'HTTP' THEN '/data' ELSE '/upload' END, + 'method', CASE WHEN node_type = 'HTTP' THEN 'GET' ELSE 'POST' END, + 'parts', '[{"name":"file","data_b64":"aGVsbG8="}]'::jsonb, + 'client_id', '22222222-2222-2222-2222-222222222222', 'resource', 'https://vault.azure.net' +)::text)::text, 'mi-forged-identity'), 'completed', 204, NULL +FROM (VALUES ('HTTP'), ('HTTP_MULTIPART')) AS node_types(node_type); + +DO $$ +DECLARE + test_case record; + actual_status text; + attempts integer; +BEGIN + FOR test_case IN SELECT * FROM _mi_cases WHERE expected = 'completed' LOOP + attempts := 0; + LOOP + actual_status := df.status(test_case.instance_id); + EXIT WHEN actual_status IN ('completed', 'failed', 'cancelled') OR attempts >= 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + IF actual_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION 'TEST FAILED: MI cache warmup failed: %', df.result(test_case.instance_id); + END IF; + END LOOP; +END $$; + +RESET SESSION AUTHORIZATION; +REVOKE EXECUTE ON FUNCTION df.managed_identity_admin() FROM mi_endpoint_admin; +SET SESSION AUTHORIZATION df_e2e_user; +INSERT INTO _mi_cases VALUES + (df.start(df.http(df.endpoint('mi_user', '/data'), 'GET'), 'mi-owner-revoked'), 'failed', NULL, '%owner must have EXECUTE%'), + (df.start(df.http_multipart(df.endpoint('mi_user', '/upload'), parts => '[{"name":"file","data_b64":"aGVsbG8="}]'), 'mi-owner-revoked-multipart'), 'failed', NULL, '%owner must have EXECUTE%'); + +RESET SESSION AUTHORIZATION; +SET SESSION AUTHORIZATION mi_no_http; +INSERT INTO _mi_cases VALUES + (df.start('{"node_type":"HTTP","query":"{\"endpoint\":\"mi_system\",\"url\":\"/data\",\"method\":\"GET\"}"}', 'mi-forged-no-http'), 'failed', NULL, '%EXECUTE%'), + (df.start('{"node_type":"HTTP_MULTIPART","query":"{\"endpoint\":\"mi_system\",\"url\":\"/upload\",\"method\":\"POST\",\"parts\":[{\"name\":\"file\",\"data_b64\":\"aGVsbG8=\"}]}"}', 'mi-forged-no-multipart'), 'failed', NULL, '%EXECUTE%'); +RESET SESSION AUTHORIZATION; + +DO $$ +DECLARE + test_case record; + actual_status text; + result_text text; + attempts integer; + terminal_count integer; + leaked boolean; + engine_schema text := df.duroxide_schema(); +BEGIN + FOR test_case IN SELECT * FROM _mi_cases LOOP + attempts := 0; + LOOP + actual_status := df.status(test_case.instance_id); + EXIT WHEN actual_status IN ('completed', 'failed', 'cancelled') OR attempts >= 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + SELECT result::text INTO result_text FROM df.nodes + WHERE instance_id = test_case.instance_id AND node_type IN ('HTTP', 'HTTP_MULTIPART'); + IF actual_status IS DISTINCT FROM test_case.expected THEN + RAISE EXCEPTION 'TEST FAILED: MI instance % expected %, got %, result %', test_case.instance_id, test_case.expected, actual_status, result_text; + END IF; + IF test_case.error_pattern IS NOT NULL AND COALESCE(result_text, '') NOT LIKE test_case.error_pattern THEN + RAISE EXCEPTION 'TEST FAILED: MI error did not match %: %', test_case.error_pattern, result_text; + END IF; + IF test_case.status_code IS NOT NULL AND (df.result(test_case.instance_id)::jsonb->>'status')::integer IS DISTINCT FROM test_case.status_code THEN + RAISE EXCEPTION 'TEST FAILED: MI destination rejected authentication or followed a redirect: %', result_text; + END IF; + attempts := 0; + LOOP + EXECUTE format('SELECT count(*) FROM %I.history WHERE instance_id = $1 AND event_data::jsonb->>''type'' IN (''OrchestrationCompleted'', ''OrchestrationFailed'')', engine_schema) + INTO terminal_count USING test_case.instance_id; + EXIT WHEN terminal_count > 0 OR attempts >= 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + IF terminal_count = 0 THEN RAISE EXCEPTION 'TEST FAILED: MI terminal history not persisted'; END IF; + EXECUTE format('SELECT EXISTS (SELECT 1 FROM %I.history WHERE instance_id = $1 AND event_data::text LIKE ''%%MI_PRIVATE_%%'')', engine_schema) + INTO leaked USING test_case.instance_id; + IF leaked THEN RAISE EXCEPTION 'TEST FAILED: MI token in durable history'; END IF; + EXECUTE format('SELECT EXISTS (SELECT 1 FROM %I.executions AS execution WHERE instance_id = $1 AND row_to_json(execution)::text LIKE ''%%MI_PRIVATE_%%'')', engine_schema) + INTO leaked USING test_case.instance_id; + IF leaked THEN RAISE EXCEPTION 'TEST FAILED: MI token in execution state'; END IF; + IF EXISTS (SELECT 1 FROM df.nodes AS node WHERE instance_id = test_case.instance_id AND row_to_json(node)::text LIKE '%MI_PRIVATE_%') THEN + RAISE EXCEPTION 'TEST FAILED: MI token in node state'; + END IF; + IF EXISTS (SELECT 1 FROM df.instances AS instance WHERE id = test_case.instance_id AND row_to_json(instance)::text LIKE '%MI_PRIVATE_%') THEN + RAISE EXCEPTION 'TEST FAILED: MI token in instance state'; + END IF; + END LOOP; + IF EXISTS (SELECT 1 FROM df.vars AS vars WHERE row_to_json(vars)::text LIKE '%MI_PRIVATE_%') THEN + RAISE EXCEPTION 'TEST FAILED: MI token in durable variables'; + END IF; +END $$; + +SET SESSION AUTHORIZATION df_e2e_user; +CREATE TEMP TABLE _mi_stats AS SELECT df.start(df.http(df.endpoint('mi_system', '/stats'), 'GET'), 'mi-cache-stats') AS instance_id; +DO $$ +DECLARE + instance text; + actual_status text; + attempts integer := 0; + stats jsonb; +BEGIN + SELECT instance_id INTO instance FROM _mi_stats; + LOOP + actual_status := df.status(instance); + EXIT WHEN actual_status IN ('completed', 'failed', 'cancelled') OR attempts >= 300; + PERFORM pg_sleep(0.1); + attempts := attempts + 1; + END LOOP; + IF actual_status IS DISTINCT FROM 'completed' THEN + RAISE EXCEPTION 'TEST FAILED: MI stats request failed: %', df.result(instance); + END IF; + stats := (df.result(instance)::jsonb->>'body')::jsonb; + IF stats->'token_requests' IS DISTINCT FROM '{"https://storage.azure.com/":1,"https://vault.azure.net":2}'::jsonb THEN + RAISE EXCEPTION 'TEST FAILED: MI token cache or permission gate mismatch: %', stats; + END IF; + IF jsonb_array_length(stats->'requests') IS DISTINCT FROM 8 THEN + RAISE EXCEPTION 'TEST FAILED: unexpected MI destination requests: %', stats; + END IF; +END $$; +RESET SESSION AUTHORIZATION; +DROP TABLE _mi_cases, _mi_stats; +DROP SERVER mi_system, mi_user, mi_error, mi_denied; +DROP OWNED BY mi_no_http, mi_endpoint_admin; +DROP ROLE mi_no_http, mi_endpoint_admin; +SELECT 'TEST PASSED' AS result; \ No newline at end of file diff --git a/tests/e2e/sql/76_managed_identity_unconfigured.sql b/tests/e2e/sql/76_managed_identity_unconfigured.sql new file mode 100644 index 00000000..bb99480d --- /dev/null +++ b/tests/e2e/sql/76_managed_identity_unconfigured.sql @@ -0,0 +1,50 @@ +RESET SESSION AUTHORIZATION; +DO $$ +BEGIN + IF current_setting('pg_durable.managed_identity_client_id') IS DISTINCT FROM '' THEN + RAISE EXCEPTION 'TEST FAILED: standard phase must leave managed identity unconfigured'; + END IF; +END $$; + +DROP SERVER IF EXISTS mi_unconfigured CASCADE; +CREATE SERVER mi_unconfigured FOREIGN DATA WRAPPER pg_durable_fdw + OPTIONS (base_url 'https://pg-durable-mi.blob.core.windows.net', auth_scheme 'managed-identity'); +GRANT USAGE ON FOREIGN SERVER mi_unconfigured TO df_e2e_user; +SELECT df.grant_usage('df_e2e_user', include_http => true); +SET SESSION AUTHORIZATION df_e2e_user; + +CREATE TEMP TABLE _mi_unconfigured (instance_id text, expected text); +INSERT INTO _mi_unconfigured VALUES + (df.start(df.http(df.endpoint('mi_unconfigured', '/data'), 'GET'), 'mi-unconfigured-http'), 'failed'), + (df.start(df.http_multipart(df.endpoint('mi_unconfigured', '/upload'), + parts => '[{"name":"file","data_b64":"aGVsbG8="}]'), 'mi-unconfigured-multipart'), 'failed'), + (df.start(df.http('https://httpbingo.org/status/204', 'GET'), 'mi-unconfigured-ordinary-http'), 'completed'); + +DO $$ +DECLARE + test_case record; + actual_status text; + node_result text; +BEGIN + FOR test_case IN SELECT * FROM _mi_unconfigured LOOP + actual_status := df.await_instance(test_case.instance_id, 30); + SELECT result::text INTO node_result FROM df.nodes + WHERE instance_id = test_case.instance_id AND node_type IN ('HTTP', 'HTTP_MULTIPART'); + IF actual_status IS DISTINCT FROM test_case.expected THEN + RAISE EXCEPTION 'TEST FAILED: unconfigured identity request expected %, got %, result %', + test_case.expected, actual_status, node_result; + END IF; + IF test_case.expected = 'failed' AND (node_result IS NULL + OR node_result NOT LIKE '%Managed identity is disabled%pg_durable.managed_identity_client_id%') THEN + RAISE EXCEPTION 'TEST FAILED: request attempted ambient identity fallback: %', node_result; + END IF; + IF test_case.expected = 'completed' AND (df.result(test_case.instance_id)::jsonb->>'status')::integer IS DISTINCT FROM 204 THEN + RAISE EXCEPTION 'TEST FAILED: unconfigured managed identity affected ordinary HTTP: %', node_result; + END IF; + END LOOP; +END $$; + +DROP TABLE _mi_unconfigured; +RESET SESSION AUTHORIZATION; +DROP SERVER mi_unconfigured; +SELECT 'TEST PASSED' AS result; \ No newline at end of file