mirror of
https://github.com/usestrix/strix.git
synced 2026-10-15 04:18:10 +00:00
Merge e1a4d9ba1a into 87f2d783d6
This commit is contained in:
Matthew Flanagan
GitHub
commit
171500bf5c
17 changed files with 2358 additions and 35 deletions
|
|
@ -62,6 +62,46 @@ Configure Strix using environment variables or a config file.
|
|||
Token block size that providers cache prompts in. A turn counts as a cache miss in the run report only when the cached tokens fall at least this many tokens short of the previous prompt and the prompt did not shrink. Lower it for providers with smaller blocks (DeepSeek and GLM use 64; vLLM defaults to 16).
|
||||
</ParamField>
|
||||
|
||||
### Anthropic Workload Identity Federation
|
||||
|
||||
Authenticate to Anthropic with short-lived tokens from your identity provider instead of `LLM_API_KEY`, which must be unset or empty. See [Anthropic](/llm-providers/anthropic#workload-identity-federation-no-api-key) and [Anthropic via Microsoft Entra ID](/llm-providers/anthropic-entra-id).
|
||||
|
||||
<ParamField path="ANTHROPIC_FEDERATION_RULE_ID" type="string">
|
||||
Federation rule to exchange against (`fdrl_…`). Required, with `ANTHROPIC_ORGANIZATION_ID`, `ANTHROPIC_SERVICE_ACCOUNT_ID`, and an identity token source.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="ANTHROPIC_ORGANIZATION_ID" type="string">
|
||||
Anthropic organization ID (a UUID).
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="ANTHROPIC_SERVICE_ACCOUNT_ID" type="string">
|
||||
Service account the rule maps to (`svac_…`).
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="ANTHROPIC_WORKSPACE_ID" type="string">
|
||||
Workspace to act in (`wrkspc_…`). Required when the rule covers several workspaces.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="ANTHROPIC_IDENTITY_TOKEN_FILE" type="string">
|
||||
File holding the identity token (JWT), read again on every exchange. Takes precedence over the other sources.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="ANTHROPIC_IDENTITY_TOKEN_COMMAND" type="string">
|
||||
Command run on every exchange whose stdout is the JWT (Strix only). Run without a shell; use `sh -c '…'` for pipelines.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="ANTHROPIC_IDENTITY_TOKEN" type="string">
|
||||
A literal JWT, for short runs. Never saved to the config file.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_ENTRA_TENANT_ID" type="string">
|
||||
Default Entra tenant for `strix auth login entra`.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_ENTRA_APP_ID" type="string">
|
||||
Default Entra app registration for `strix auth login entra`: the app that is both the OAuth client and the token audience.
|
||||
</ParamField>
|
||||
|
||||
### Dedicated deduplication model
|
||||
|
||||
Finding deduplication is a cheap, structured classification task. By default it
|
||||
|
|
|
|||
|
|
@ -35,6 +35,7 @@
|
|||
"llm-providers/overview",
|
||||
"llm-providers/openai",
|
||||
"llm-providers/anthropic",
|
||||
"llm-providers/anthropic-entra-id",
|
||||
"llm-providers/openrouter",
|
||||
"llm-providers/vercel-ai-gateway",
|
||||
"llm-providers/vertex",
|
||||
|
|
|
|||
317
docs/llm-providers/anthropic-entra-id.mdx
Normal file
317
docs/llm-providers/anthropic-entra-id.mdx
Normal file
|
|
@ -0,0 +1,317 @@
|
|||
---
|
||||
title: "Anthropic via Microsoft Entra ID"
|
||||
description: "Run Strix on Claude with no API key, federating Microsoft Entra ID identities to Anthropic"
|
||||
---
|
||||
|
||||
Strix can call Claude without an `LLM_API_KEY`. It exchanges a short-lived Microsoft Entra ID token for a short-lived Anthropic access token (`sk-ant-oat01-…`) through Anthropic's [Workload Identity Federation](https://platform.claude.com/docs/en/manage-claude/workload-identity-federation), and repeats the exchange before it expires, so long scans keep running. No static Anthropic key exists anywhere.
|
||||
|
||||
This guide covers two kinds of host:
|
||||
|
||||
| Host | Entra identity | Token source |
|
||||
|---|---|---|
|
||||
| Azure VM | The VM's managed identity | `ANTHROPIC_IDENTITY_TOKEN_COMMAND` querying the instance metadata service (IMDS) |
|
||||
| Any other Linux host: a laptop or desktop, a local VM, or a VM on-premises or in another public or private cloud | Your Entra user | `strix auth login entra` |
|
||||
|
||||
<Note>
|
||||
Federation applies only to `anthropic/…` models, which call Anthropic directly. **An API key always wins:** if `LLM_API_KEY` is set, in the environment or in the Strix config file, Strix uses it and prints a warning. Federated runs record `"auth_mode": "workload_identity"` in `strix_runs/<run>/run.json`.
|
||||
</Note>
|
||||
|
||||
## How setup fits together
|
||||
|
||||
Setup is done once; after that you just run scans.
|
||||
|
||||
| Step | Who | Where | How often |
|
||||
|---|---|---|---|
|
||||
| [1. Entra ID setup](#1-entra-id-setup) | Entra admin | Admin workstation | Once per tenant |
|
||||
| [2. Prepare the identity](#2-prepare-the-identity) | The user, or the VM's owner | The Strix host | Once per user, or per VM |
|
||||
| [3. Claude Console setup](#3-claude-console-setup) | Anthropic org admin | Claude Console | Issuer once per tenant; a rule and service account per user or VM |
|
||||
| [4. Configure Strix and test](#4-configure-strix-and-test) | The user | The Strix host | Once per host |
|
||||
| [5. Everyday use](#5-everyday-use) | The user | The Strix host | Every scan |
|
||||
|
||||
Step 2 comes before the Console because the federation rule matches claims (`oid`, `sub`) that you can only read from the identity's token. Do steps 3 and 4 in one sitting: running `strix auth test` within 15 minutes of finishing the Console wizard marks the connection as verified there.
|
||||
|
||||
**Prerequisites:** Strix and Docker on the Strix host (plus `curl` and `jq` on an Azure VM), and outbound HTTPS from it to `login.microsoftonline.com` and `api.anthropic.com`.
|
||||
|
||||
**Placeholders:**
|
||||
|
||||
| Placeholder | Value | Found in |
|
||||
|---|---|---|
|
||||
| `<TENANT_ID>` | Your Entra tenant ID | Entra ID → Overview, or step 1.1 |
|
||||
| `<APP_ID>` | Client ID of the `claude-api-federation` app registration | Step 1.1 |
|
||||
| `<USER_OID>`, `<USER_SUB>` | The user's `oid` and `sub` claims | Step 2, Entra user |
|
||||
| `<VM_MI_OID>` | The VM managed identity's object ID | Step 2, Azure VM |
|
||||
|
||||
## 1. Entra ID setup
|
||||
|
||||
Once per tenant. You need an Entra role that can create app registrations. Run these from an admin workstation.
|
||||
|
||||
<Warning>
|
||||
A common Conditional Access policy, "Azure management – limit to administrative roles", blocks `az login` for non-admins, because `az login` always requests an Azure management token. Nothing at runtime needs Azure management; only these setup steps (and enabling a VM's managed identity in step 2) use `az`.
|
||||
</Warning>
|
||||
|
||||
<Steps>
|
||||
<Step title="Create the audience app registration">
|
||||
Entra only issues tokens for an audience that exists in your tenant. This app is the "Claude API" audience. It has no secrets and no permissions.
|
||||
|
||||
```bash
|
||||
az login
|
||||
TENANT_ID=$(az account show --query tenantId -o tsv)
|
||||
APP_ID=$(az ad app create --display-name claude-api-federation --query appId -o tsv)
|
||||
OBJ_ID=$(az ad app show --id "$APP_ID" --query id -o tsv)
|
||||
```
|
||||
</Step>
|
||||
|
||||
<Step title="Set the identifier URI and v2.0 tokens">
|
||||
`az ad app update --set api.requestedAccessTokenVersion=2` fails on a new app (`Couldn't find 'api' in ''`), so patch through Microsoft Graph:
|
||||
|
||||
```bash
|
||||
az rest --method PATCH \
|
||||
--uri "https://graph.microsoft.com/v1.0/applications/$OBJ_ID" \
|
||||
--headers 'Content-Type=application/json' \
|
||||
--body "{\"identifierUris\":[\"api://$APP_ID\"],\"api\":{\"requestedAccessTokenVersion\":2}}"
|
||||
```
|
||||
|
||||
v2.0 tokens carry `iss` = `https://login.microsoftonline.com/<TENANT_ID>/v2.0` and `aud` = `<APP_ID>` (the bare GUID).
|
||||
</Step>
|
||||
|
||||
<Step title="Create the service principal">
|
||||
```bash
|
||||
az ad sp create --id "$APP_ID"
|
||||
```
|
||||
|
||||
The output should show `accountEnabled: true`, `servicePrincipalNames` including `api://<APP_ID>`, and `appRoleAssignmentRequired: false`. Any identity in the tenant can then get a token for this audience; the Anthropic federation rule decides who is accepted.
|
||||
</Step>
|
||||
|
||||
<Step title="For user sign-in: expose a scope and make the app its own public client">
|
||||
`strix auth login entra` signs in **as this app** (`client_id` = `<APP_ID>`), so the app is both the OAuth client and the API. It needs a delegated scope and a `http://localhost` public-client redirect. An app requesting its own scope needs no consent, so users see no prompt and no admin grant is required. Skip this step if you only use managed identities.
|
||||
|
||||
```bash
|
||||
SCOPE_ID=$(uuidgen)
|
||||
az rest --method PATCH \
|
||||
--uri "https://graph.microsoft.com/v1.0/applications/$OBJ_ID" \
|
||||
--headers 'Content-Type=application/json' \
|
||||
--body "{\"api\":{\"requestedAccessTokenVersion\":2,\"oauth2PermissionScopes\":[{
|
||||
\"id\":\"$SCOPE_ID\",\"value\":\"user_impersonation\",\"type\":\"User\",\"isEnabled\":true,
|
||||
\"adminConsentDisplayName\":\"Access Claude API federation\",
|
||||
\"adminConsentDescription\":\"Request tokens for Anthropic workload identity federation\",
|
||||
\"userConsentDisplayName\":\"Access Claude API federation\",
|
||||
\"userConsentDescription\":\"Request tokens for Anthropic workload identity federation\"}]}}"
|
||||
|
||||
az rest --method PATCH \
|
||||
--uri "https://graph.microsoft.com/v1.0/applications/$OBJ_ID" \
|
||||
--headers 'Content-Type=application/json' \
|
||||
--body '{"publicClient":{"redirectUris":["http://localhost"]}}'
|
||||
|
||||
az ad app show --id "$APP_ID" --query '{ver:api.requestedAccessTokenVersion, scopes:api.oauth2PermissionScopes[].value, redirects:publicClient.redirectUris, preauth:api.preAuthorizedApplications[].appId}'
|
||||
# → ver: 2, scopes: ["user_impersonation"], redirects: ["http://localhost"], preauth: []
|
||||
```
|
||||
|
||||
Keep `preAuthorizedApplications` empty. Using this app, rather than a client such as the Azure CLI, keeps the stored refresh token limited to this audience (see [Security](#security)).
|
||||
</Step>
|
||||
</Steps>
|
||||
|
||||
## 2. Prepare the identity
|
||||
|
||||
Once per user, or per VM. This gets the identity working on its Strix host and gives you the claims the Console rule needs.
|
||||
|
||||
<Tabs>
|
||||
<Tab title="Entra user (any other host)">
|
||||
On the Strix host, sign in. This needs only step 1, no Anthropic settings yet.
|
||||
|
||||
```bash
|
||||
export STRIX_ENTRA_TENANT_ID="<TENANT_ID>"
|
||||
export STRIX_ENTRA_APP_ID="<APP_ID>"
|
||||
strix auth login entra # over SSH: strix auth login entra --manual
|
||||
```
|
||||
|
||||
Strix prints a sign-in URL. Open it in a browser and sign in.
|
||||
|
||||
- **On the host itself,** the browser returns to a local callback on port 8400 and the sign-in completes.
|
||||
- **Over SSH,** use `--manual`. Open the URL in your workstation's browser; after sign-in it lands on a `http://localhost:8400/?code=…` page that won't load. Copy that URL from the address bar and paste it into Strix. Alternatively, tunnel the port (`ssh -L 8400:127.0.0.1:8400 <user>@<strix-host>`) and run without `--manual`.
|
||||
|
||||
Sign-in happens in your browser, so device-based Conditional Access is evaluated there. Strix stores the sign-in in `~/.strix/entra-auth.json` and refreshes the Entra token from the host from then on.
|
||||
|
||||
Read the claims:
|
||||
|
||||
```bash
|
||||
strix auth status
|
||||
# Microsoft Entra ID (Anthropic Workload Identity Federation)
|
||||
# ...
|
||||
# aud: OK
|
||||
# azp: <APP_ID>
|
||||
# oid: <USER_OID>
|
||||
# sub: <USER_SUB>
|
||||
```
|
||||
|
||||
Check that `aud` shows `OK` and `azp` is `<APP_ID>`. `oid` is the user's object ID (also on **Entra ID → Users → (user) → Object ID**). `sub` appears only in tokens: it is specific to the user *and* the app, so it has to come from a token for `<APP_ID>`, as here. Neither changes across sign-ins, token refreshes or password changes, or between hosts, so one rule covers the user wherever they sign in.
|
||||
|
||||
<Tip>
|
||||
`STRIX_ENTRA_TENANT_ID` and `STRIX_ENTRA_APP_ID` are the defaults for `strix auth login entra` (or pass `--tenant` and `--app`). The sign-in remembers its tenant and app; if those variables later name a different one, the stored sign-in isn't used.
|
||||
</Tip>
|
||||
</Tab>
|
||||
|
||||
<Tab title="Azure VM">
|
||||
No sign-in and no secrets: the VM gets its Entra token from IMDS. Steps 1.1–1.3 are required; 1.4 isn't.
|
||||
|
||||
From the admin workstation, enable the VM's system-assigned managed identity; the command prints its object ID:
|
||||
|
||||
```bash
|
||||
az vm identity assign -g <RG> -n <VM> --query systemAssignedIdentity -o tsv # → <VM_MI_OID>
|
||||
```
|
||||
|
||||
On the VM, check that it gets a token for the app:
|
||||
|
||||
```bash
|
||||
curl -sSf -H 'Metadata: true' \
|
||||
"http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=api://<APP_ID>" \
|
||||
| jq -r .access_token \
|
||||
| jq -R 'split(".")[1] | gsub("-";"+") | gsub("_";"/") | @base64d | fromjson | {ver, aud, oid, sub}'
|
||||
```
|
||||
|
||||
Expect `ver` 2.0, `aud` = `<APP_ID>`, and `oid` = `sub` = `<VM_MI_OID>`: in a managed identity token the subject is the object ID, so there's no separate value to look up.
|
||||
|
||||
With a user-assigned identity, add `&client_id=<IDENTITY_CLIENT_ID>` to the IMDS URL, here and in step 4. Any process on the VM can request this identity's token, which is why the rule pins the managed identity's `oid` and the VM gets its own service account.
|
||||
</Tab>
|
||||
</Tabs>
|
||||
|
||||
## 3. Claude Console setup
|
||||
|
||||
You need the admin, owner or primary owner role in the Anthropic organization. All workloads in a tenant share **one issuer**, created by the first run of the wizard. Create **one federation rule and one service account per user or VM**, so each can be audited and revoked on its own.
|
||||
|
||||
<Steps>
|
||||
<Step title="Run the Connect workload wizard">
|
||||
Open **Settings → Workload identity → Connect workload → Microsoft Entra ID**.
|
||||
|
||||
- **Token issuer:** choose **v2.0 (login.microsoftonline.com)**; the default, v1, is wrong for this setup. The issuer URL is `https://login.microsoftonline.com/<TENANT_ID>/v2.0`. Click **Verify issuer**.
|
||||
- **Match conditions:**
|
||||
|
||||
| Field | Entra user | Azure VM |
|
||||
|---|---|---|
|
||||
| **Object (principal) ID** (required) | `<USER_OID>` | `<VM_MI_OID>` |
|
||||
| **Application (client) ID** | `<APP_ID>` | `<APP_ID>` |
|
||||
| Advanced → **Subject pattern** | `<USER_SUB>` (exact, no `*`) | leave Advanced untouched |
|
||||
| Advanced → **Expected audience** | `<APP_ID>` | — |
|
||||
| Advanced → claims `oid` / `tid` / `azp` | `<USER_OID>` / `<TENANT_ID>` / `<APP_ID>` | — |
|
||||
|
||||
- **Service account:** **Create a new one**, named after the workload (for example `strix-<host>-<user>` or `strix-azure-vm`), in the workspace whose rate limits and billing Strix should use. Scope `workspace:developer`, token lifetime 600 s.
|
||||
|
||||
Record the `federation_rule_id` (`fdrl_…`), `service_account_id` (`svac_…`), `organization_id` and `workspace_id` (`wrkspc_…`) shown in the wizard's test snippet; step 4 needs them. Don't run the snippet itself: it always queries IMDS, and step 4 tests the same exchange on any host with `strix auth test`.
|
||||
</Step>
|
||||
|
||||
<Step title="Raise the issuer's JWT lifetime (first VM only)">
|
||||
The wizard creates the issuer with `max_jwt_lifetime_seconds = 7500`. Managed identity tokens live up to 24 hours, so VM exchanges fail with an opaque `401` until you raise it: **Settings → Workload identity → Issuers → (issuer) → Edit → `max_jwt_lifetime_seconds` = `86400`**. This is an issuer setting, so do it once. User tokens (60–90 minutes) work with either value.
|
||||
</Step>
|
||||
</Steps>
|
||||
|
||||
<Note>
|
||||
Why the Entra user rule needs **Advanced match options**: the guided fields assume a managed identity, where `sub` equals `oid`. In a user token, `sub` is a different opaque value, so set **Subject pattern** to it explicitly. Once you edit the Advanced section it replaces the guided fields, so **Expected audience** must be filled in; blank means "the default Anthropic API audience", which Entra tokens never carry. A user's `sub` and `oid` don't change across sign-ins, token refreshes or password changes. Never use a wildcard or partial value: every identity in the tenant can get a token for `<APP_ID>`, so the `oid` match is what restricts access.
|
||||
</Note>
|
||||
|
||||
## 4. Configure Strix and test
|
||||
|
||||
Once per host. On the Strix host, set the IDs from step 3, with no API key:
|
||||
|
||||
```bash
|
||||
export STRIX_LLM="anthropic/claude-sonnet-4-6"
|
||||
export ANTHROPIC_ORGANIZATION_ID="<org uuid>"
|
||||
export ANTHROPIC_WORKSPACE_ID="wrkspc_..."
|
||||
export ANTHROPIC_FEDERATION_RULE_ID="fdrl_..."
|
||||
export ANTHROPIC_SERVICE_ACCOUNT_ID="svac_..."
|
||||
export LLM_API_KEY= # empty, not unset: also clears a key saved in the Strix config
|
||||
```
|
||||
|
||||
On an Azure VM, also tell Strix how to get the managed identity's token. An Entra user needs nothing more: the sign-in from step 2 is the token source.
|
||||
|
||||
```bash
|
||||
export ANTHROPIC_IDENTITY_TOKEN_COMMAND="sh -c 'curl -sSf -H Metadata:true \"http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=api://<APP_ID>\" | jq -r .access_token'"
|
||||
```
|
||||
|
||||
Test the exchange with the same settings and identity token source a scan would use:
|
||||
|
||||
```bash
|
||||
strix auth test
|
||||
# Anthropic Workload Identity Federation
|
||||
# Identity token source: Entra ID sign-in (strix auth login entra)
|
||||
# iss: https://login.microsoftonline.com/<TENANT_ID>/v2.0
|
||||
# aud: <APP_ID>
|
||||
# azp: <APP_ID>
|
||||
# oid: <USER_OID>
|
||||
# sub: <USER_SUB>
|
||||
# Exchange OK: sk-ant-oat01-… valid for 600s
|
||||
```
|
||||
|
||||
On a VM the source is `ANTHROPIC_IDENTITY_TOKEN_COMMAND` and `oid` = `sub` = `<VM_MI_OID>`. `strix auth test` exits `0` on success and `1` otherwise, printing why: a missing setting, the token source's error, or Anthropic's error with its `request-id` (the reason for a `401` is in the Console's authentication history; see [Troubleshooting](#troubleshooting)). It never prints a token. It also warns when `LLM_API_KEY` or a non-Anthropic `STRIX_LLM` would stop scans using federation.
|
||||
|
||||
Then run a first scan from the same shell:
|
||||
|
||||
```bash
|
||||
strix -n -t <target> --scan-mode quick
|
||||
```
|
||||
|
||||
When a scan starts, Strix saves these settings (including `STRIX_ENTRA_*` and the token command) to its config file, `~/.strix/cli-config.json`, and deletes any saved `LLM_API_KEY`, because you set it to empty. Later shells don't need the exports. If you run Strix with `--config PATH`, the settings go to that file instead; pass the same `--config PATH` to `strix auth test`.
|
||||
|
||||
## 5. Everyday use
|
||||
|
||||
Run scans as usual:
|
||||
|
||||
```bash
|
||||
strix -n -t <target>
|
||||
```
|
||||
|
||||
Strix exchanges a token at startup, and again roughly every 8 minutes during long scans, refreshing the Entra token as needed. A healthy run:
|
||||
|
||||
- exits immediately with *"Anthropic workload identity federation failed: …"* if the setup is wrong, before any scanning starts;
|
||||
- prints no *"LLM_API_KEY is set, so it is used instead of …"* warning;
|
||||
- records `"auth_mode": "workload_identity"` in `run.json`;
|
||||
- shows successful exchanges for your rule in **Console → Settings → Workload identity → Authentication history**.
|
||||
|
||||
| When | Do |
|
||||
|---|---|
|
||||
| A scan fails with *"Entra ID sign-in failed (…); run: strix auth login entra"* | Sign in again with `strix auth login entra` (`--manual` over SSH). The refresh token expires after about 90 days unused, or sooner under a tenant sign-in frequency policy. |
|
||||
| You want to check the setup before a long scan | `strix auth test` |
|
||||
| You want to see who is signed in on this host | `strix auth status` |
|
||||
| You stop using this host | `strix auth logout` removes the stored sign-ins (ChatGPT and Entra). It doesn't revoke the Entra session; revoke the user's sessions in Entra if the host may be compromised. |
|
||||
| A user leaves, or a VM is retired | Disable or delete their federation rule and service account in the Console. |
|
||||
| You add another user or VM | Repeat steps 2–4 for it. Step 1 and the issuer are reused. |
|
||||
|
||||
## Security
|
||||
|
||||
| Asset | Where | If it leaks |
|
||||
|---|---|---|
|
||||
| Entra refresh token (user sign-in) | `~/.strix/entra-auth.json`, mode `600` | Mints Claude-audience Entra tokens as the user for up to ~90 days. Revoke the user's sessions in Entra, or disable the federation rule. |
|
||||
| Entra access token (aud = `<APP_ID>`) | The same file, and Strix's memory | Exchangeable at Anthropic until it expires (60–90 minutes; Entra tokens carry no `jti`). Disable the federation rule. |
|
||||
| Anthropic access token (`sk-ant-oat01-…`) | Strix's memory only | Usable for at most the rule's token lifetime (600 s). |
|
||||
|
||||
**The stored refresh token only works for the Claude audience.** A refresh token can be redeemed for whatever its client app may access, so the client decides what a stolen token is worth. `strix auth login entra` signs in as `claude-api-federation` itself, not as a broad first-party client such as the Azure CLI, whose refresh tokens reach Microsoft Graph, Azure Resource Manager and more. Redeemed directly at Entra, this refresh token gets:
|
||||
|
||||
| Redeemed for | Result |
|
||||
|---|---|
|
||||
| `api://<APP_ID>/user_impersonation` | Token issued |
|
||||
| Microsoft Graph | `AADSTS65001`: the app has no consent for Graph |
|
||||
| Azure Resource Manager | Refused: no consent (and Conditional Access, where it applies) |
|
||||
| Any resource, as another client | `AADSTS70000`: the refresh token is bound to `<APP_ID>` |
|
||||
|
||||
Keep it that way: don't add API permissions or consent grants to the app, and keep `preAuthorizedApplications` empty. For an Entra-side allow-list of users, use a separate client app registration with "Assignment required"; turning that on for `claude-api-federation` would also block VMs' managed identities unless they get an app role assignment.
|
||||
|
||||
The sign-in callback listens on `127.0.0.1:8400` only and uses PKCE and `state`. The Docker sandbox receives none of the `ANTHROPIC_*` settings, and token commands run on the host. **Don't use your home directory as a scan target**, which would copy `~/.strix` into the sandbox.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
The token endpoint only ever returns `401 Authentication failed`. **The reason is on the Console's Settings → Workload identity → Authentication history page.**
|
||||
|
||||
| Symptom | Cause | Fix |
|
||||
|---|---|---|
|
||||
| `AADSTS53003` from `az login` | Conditional Access blocks Azure management for non-admins | Expected; use `strix auth login entra` |
|
||||
| `AADSTS50011` (redirect URI mismatch) on `strix auth login entra` | No `http://localhost` public-client redirect | Step 1.4 |
|
||||
| `AADSTS65001` on `strix auth login entra` | Wrong `--app`, or the scope is missing | Step 1.4; check `STRIX_ENTRA_APP_ID` |
|
||||
| `AADSTS500011` / `AADSTS50001` | Audience app or its service principal missing | Steps 1.1–1.3 |
|
||||
| Over SSH, sign-in waits and then asks for a URL | The browser can't reach the host's port 8400 | Use `--manual` and paste the redirect URL, or tunnel the port |
|
||||
| Token `ver` is `"1.0"` | v2.0 setting didn't apply | Step 1.2 |
|
||||
| `401`, history: audience mismatch | **Expected audience** blank or set to `api://…` | Set it to the bare `<APP_ID>` GUID |
|
||||
| `401`, history: subject mismatch | Rule subject is the `oid`, but user tokens have a different `sub` | Set **Subject pattern** to the `sub` from `strix auth status` |
|
||||
| `401`, history: claim mismatch on `azp` | Rule's `azp` isn't `<APP_ID>` | Set it to `<APP_ID>` |
|
||||
| `401`, history: lifetime (VM) | Issuer still at 7500 s | Raise `max_jwt_lifetime_seconds` to `86400` |
|
||||
| `Entra ID sign-in failed (invalid_grant: …); run: strix auth login entra` | Refresh token expired or revoked | Sign in again |
|
||||
| Strix warns *"LLM_API_KEY is set…"* | A key in the environment or the Strix config file | `export LLM_API_KEY=` (empty) |
|
||||
|
||||
See also: the [Anthropic provider page](/llm-providers/anthropic) for the federation variables and other identity providers, and Anthropic's [WIF with Microsoft Entra ID](https://platform.claude.com/docs/en/manage-claude/wif-providers/azure) guide.
|
||||
|
|
@ -22,3 +22,85 @@ export LLM_API_KEY="sk-ant-..."
|
|||
1. Go to [console.anthropic.com](https://console.anthropic.com)
|
||||
2. Navigate to API Keys
|
||||
3. Create a new key
|
||||
|
||||
## Workload Identity Federation (no API key)
|
||||
|
||||
Strix can authenticate to Anthropic with [Workload Identity Federation](https://platform.claude.com/docs/en/manage-claude/workload-identity-federation) instead of a static key. It exchanges a JWT from your identity provider for a short-lived access token and re-exchanges it before it expires, so long scans keep running. The token is held in memory only.
|
||||
|
||||
Set the same variables the Anthropic SDKs read, and set `LLM_API_KEY` to empty (an API key always wins, including one saved in the Strix config file; an empty value also deletes that):
|
||||
|
||||
```bash
|
||||
export STRIX_LLM="anthropic/claude-sonnet-4-6"
|
||||
export ANTHROPIC_FEDERATION_RULE_ID="fdrl_..."
|
||||
export ANTHROPIC_ORGANIZATION_ID="00000000-0000-0000-0000-000000000000"
|
||||
export ANTHROPIC_SERVICE_ACCOUNT_ID="svac_..."
|
||||
export ANTHROPIC_WORKSPACE_ID="wrkspc_..." # required if the rule covers several workspaces
|
||||
export ANTHROPIC_IDENTITY_TOKEN_FILE="/var/run/secrets/anthropic.com/token"
|
||||
export LLM_API_KEY=
|
||||
```
|
||||
|
||||
Then check the setup with `strix auth test`: it runs one exchange with the same settings and token source a scan would use, and reports the source, the identity token's claims and the result, never a token. Like other settings, these are saved to the Strix config file when a scan starts.
|
||||
|
||||
Strix re-reads `ANTHROPIC_IDENTITY_TOKEN_FILE` on every exchange, so the file must hold an unexpired JWT each time. `ANTHROPIC_IDENTITY_TOKEN` (a literal JWT) also works for short runs.
|
||||
|
||||
Federation applies only to models that call Anthropic's API directly (`anthropic/...`). A bare `claude-...` name isn't routed to Anthropic, so it doesn't federate either. Claude through Bedrock, Vertex AI, or OpenRouter keeps that provider's authentication.
|
||||
|
||||
### Other identity providers
|
||||
|
||||
Strix doesn't care which identity provider signed the JWT. Anthropic checks the issuer, signature, and claims against your federation rule. Strix takes the JWT from the first of these that is set:
|
||||
|
||||
1. `ANTHROPIC_IDENTITY_TOKEN_FILE`: a file, read again on every exchange.
|
||||
2. `ANTHROPIC_IDENTITY_TOKEN_COMMAND` (Strix only, not an SDK variable): a command run on every exchange. It must print a single JWT on stdout.
|
||||
3. `ANTHROPIC_IDENTITY_TOKEN`: a literal JWT.
|
||||
4. A Microsoft Entra ID user sign-in made with `strix auth login entra` (see [Anthropic via Microsoft Entra ID](/llm-providers/anthropic-entra-id)).
|
||||
5. The GitHub Actions OIDC token, when running in a job with `id-token: write`.
|
||||
|
||||
| Where the token comes from | Configure |
|
||||
|---|---|
|
||||
| Kubernetes, EKS, or AKS projected service-account token; SPIFFE JWT-SVID file | `ANTHROPIC_IDENTITY_TOKEN_FILE` = the mounted token path |
|
||||
| Google Cloud | `ANTHROPIC_IDENTITY_TOKEN_COMMAND="gcloud auth print-identity-token --audiences=https://api.anthropic.com"` |
|
||||
| Microsoft Entra ID user, on any host | `strix auth login entra` ([guide](/llm-providers/anthropic-entra-id)) |
|
||||
| Azure VM managed identity (IMDS) | `ANTHROPIC_IDENTITY_TOKEN_COMMAND`, see the example below and the [Entra ID guide](/llm-providers/anthropic-entra-id) |
|
||||
| Any CLI or script that prints a JWT | `ANTHROPIC_IDENTITY_TOKEN_COMMAND="/path/to/issue-token --flag"` |
|
||||
| GitHub Actions | Nothing extra (see below) |
|
||||
|
||||
On an Azure VM, where `<APP_ID>` is the app registration your federation rule expects as the audience:
|
||||
|
||||
```bash
|
||||
export ANTHROPIC_IDENTITY_TOKEN_COMMAND="sh -c 'curl -sSf -H Metadata:true \"http://169.254.169.254/metadata/identity/oauth2/token?api-version=2018-02-01&resource=api://<APP_ID>\" | jq -r .access_token'"
|
||||
```
|
||||
|
||||
How the command runs:
|
||||
|
||||
- It's split like a shell command line, but run **without** a shell. For pipes or redirection, wrap it in `sh -c '…'`, as in the Azure example.
|
||||
- A leading `~` in the program path is expanded. It has no stdin and 30 seconds to finish.
|
||||
- It inherits Strix's environment. Under `uv run`, that environment puts Strix's virtualenv first on `PATH`, so give a script its interpreter explicitly (for example `#!/usr/bin/python3` rather than `#!/usr/bin/env python3`) if it needs packages from outside that virtualenv.
|
||||
- On failure, Strix reports the exit status and the last line of stderr, never stdout.
|
||||
- Like the other `ANTHROPIC_*` settings, it's saved to the Strix config file when a scan starts and reused by later scans. To stop using it, run once with `ANTHROPIC_IDENTITY_TOKEN_COMMAND=` (empty), or remove it from the config file. The literal `ANTHROPIC_IDENTITY_TOKEN` is never saved.
|
||||
|
||||
**Single-use JWTs (`jti`).** By default Anthropic accepts a JWT that carries a `jti` claim only once. Strix exchanges again about 2 minutes before each Anthropic token expires (about every 8 minutes with a 600-second rule lifetime), and each exchange must present a JWT it hasn't presented before. A token command gets a fresh token on every exchange. A token file must have rotated by then, or the exchange fails with `jti_reused` in the Console's authentication history. Kubernetes rotates projected tokens well before they expire, but not necessarily every few minutes. If a file source can't keep up, use a token command, give the rule a longer `token_lifetime_seconds`, or, as a last resort, disable `check_jti` on that issuer. Entra ID tokens carry no `jti`, so they're not affected.
|
||||
|
||||
### GitHub Actions
|
||||
|
||||
Inside a job with `id-token: write` and no token file configured, Strix fetches a fresh GitHub OIDC token (audience `https://api.anthropic.com`) for each exchange:
|
||||
|
||||
```yaml
|
||||
permissions:
|
||||
id-token: write
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
strix:
|
||||
runs-on: ubuntu-latest
|
||||
env:
|
||||
STRIX_LLM: anthropic/claude-sonnet-4-6
|
||||
ANTHROPIC_FEDERATION_RULE_ID: fdrl_...
|
||||
ANTHROPIC_ORGANIZATION_ID: 00000000-0000-0000-0000-000000000000
|
||||
ANTHROPIC_SERVICE_ACCOUNT_ID: svac_...
|
||||
ANTHROPIC_WORKSPACE_ID: wrkspc_...
|
||||
steps:
|
||||
- uses: actions/checkout@v5
|
||||
- run: strix -n -t ./ --scan-mode quick
|
||||
```
|
||||
|
||||
`run.json` records `auth_mode: workload_identity` for federated runs. If an exchange is denied, the response is an opaque `401`; the reason is listed on the Console's **Settings → Workload identity → Authentication history** page.
|
||||
|
|
|
|||
|
|
@ -168,6 +168,20 @@ strix --target ./my-project --workspace-file ./wordlist.txt
|
|||
strix --target https://app.com --workspace-file ./openapi.yaml:specs/openapi.yaml
|
||||
```
|
||||
|
||||
## Sign-in (`strix auth`)
|
||||
|
||||
`strix auth` stores sign-ins that replace an API key.
|
||||
|
||||
| Command | What it does |
|
||||
|---------|--------------|
|
||||
| `strix auth login chatgpt [--manual]` | Sign in with a ChatGPT subscription; a `chatgpt/…` `STRIX_LLM` then runs on it |
|
||||
| `strix auth login entra [--manual] [--tenant ID] [--app ID]` | Sign in as a Microsoft Entra ID user for [Anthropic Workload Identity Federation](/llm-providers/anthropic-entra-id) |
|
||||
| `strix auth status` | Show every provider's sign-in; exits `0` if any is signed in |
|
||||
| `strix auth logout` | Remove every stored sign-in from this host |
|
||||
| `strix auth test [--config PATH]` | Run one Anthropic federation exchange with the token source a scan would use, and report the result (never prints a token) |
|
||||
|
||||
`--manual` skips the local callback server: open the printed URL in any browser, sign in, and paste the URL it redirects to back into Strix. Use it over SSH.
|
||||
|
||||
## Exit Codes
|
||||
|
||||
| Code | Meaning |
|
||||
|
|
|
|||
|
|
@ -230,6 +230,7 @@ ignore = [
|
|||
"tests/test_viewer_auth.py" = ["S105", "S106", "ARG001"]
|
||||
"tests/test_cloud_cli.py" = ["S105", "ARG001"]
|
||||
"tests/test_codex_auth.py" = ["S105", "S106", "SLF001"]
|
||||
"tests/test_entra_user.py" = ["S105", "S106", "SLF001"]
|
||||
# Hatchling loads the build hook by path, not as an importable package.
|
||||
"scripts/tui_sidecar_hook.py" = ["INP001"]
|
||||
# Stdlib HTTP handler overrides (do_GET/do_POST).
|
||||
|
|
@ -312,6 +313,8 @@ ignore = [
|
|||
# Heavy inference deps (httpx, openai) imported lazily so auth-status checks
|
||||
# don't pull them in.
|
||||
"strix/config/codex.py" = ["PLC0415"]
|
||||
# fcntl (cross-process refresh lock) is POSIX-only, so it is imported where used.
|
||||
"strix/config/entra_user.py" = ["PLC0415"]
|
||||
# Interface utility branches per scope-mode / target-type combination;
|
||||
# splitting would obscure the decision tree without simplifying it.
|
||||
"strix/interface/utils.py" = ["PLR0912", "BLE001", "PLC0415"]
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ from strix.config.loader import (
|
|||
persist_current,
|
||||
)
|
||||
from strix.config.settings import (
|
||||
AnthropicWifSettings,
|
||||
ContextSettings,
|
||||
DedupeSettings,
|
||||
IntegrationSettings,
|
||||
|
|
@ -28,6 +29,7 @@ from strix.config.settings import (
|
|||
|
||||
|
||||
__all__ = [
|
||||
"AnthropicWifSettings",
|
||||
"ContextSettings",
|
||||
"DedupeSettings",
|
||||
"IntegrationSettings",
|
||||
|
|
|
|||
319
strix/config/anthropic_wif.py
Normal file
319
strix/config/anthropic_wif.py
Normal file
|
|
@ -0,0 +1,319 @@
|
|||
"""Anthropic Workload Identity Federation: trade an IdP-issued JWT for a
|
||||
short-lived Anthropic access token, and keep it fresh for the whole scan.
|
||||
|
||||
Configured with the same env vars the Anthropic SDKs read
|
||||
(``ANTHROPIC_FEDERATION_RULE_ID``, ``ANTHROPIC_ORGANIZATION_ID``,
|
||||
``ANTHROPIC_SERVICE_ACCOUNT_ID``, ``ANTHROPIC_WORKSPACE_ID``, and
|
||||
``ANTHROPIC_IDENTITY_TOKEN_FILE`` or ``ANTHROPIC_IDENTITY_TOKEN``), so one
|
||||
environment serves Strix and any SDK workload alongside it. Strix also accepts
|
||||
``ANTHROPIC_IDENTITY_TOKEN_COMMAND``, a command whose stdout is the JWT, for IdPs
|
||||
that hand out tokens from a local endpoint or CLI rather than a file, and a
|
||||
Microsoft Entra ID user sign-in made with ``strix auth login entra``. Inside a
|
||||
GitHub Actions job with ``id-token: write`` and no token configured, the JWT is
|
||||
fetched from the runner's OIDC endpoint instead.
|
||||
|
||||
The minted ``sk-ant-oat01-`` token lives in memory only; it is never persisted.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import shlex
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
import requests
|
||||
|
||||
from strix.config import entra_user
|
||||
from strix.config.loader import load_settings
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
DEFAULT_BASE_URL = "https://api.anthropic.com"
|
||||
TOKEN_PATH = "/v1/oauth/token" # noqa: S105 # nosec B105 - URL path, not a secret
|
||||
GRANT_TYPE = "urn:ietf:params:oauth:grant-type:jwt-bearer"
|
||||
GITHUB_AUDIENCE = "https://api.anthropic.com"
|
||||
|
||||
_TIMEOUT_S = 30
|
||||
# The Anthropic SDKs' refresh schedule: try early, and only fail once the cached
|
||||
# token is too close to expiry to send.
|
||||
_ADVISORY_REFRESH_S = 120
|
||||
_MANDATORY_REFRESH_S = 30
|
||||
|
||||
_lock = threading.Lock()
|
||||
_cached: tuple[str, float] | None = None # (access token, expires-at epoch seconds)
|
||||
|
||||
|
||||
class WifAuthError(RuntimeError):
|
||||
"""The federated token exchange (or fetching the JWT for it) failed."""
|
||||
|
||||
|
||||
def is_configured() -> bool:
|
||||
"""Every field the exchange needs is set, and a JWT source exists."""
|
||||
wif = load_settings().anthropic_wif
|
||||
ids = (wif.federation_rule_id, wif.organization_id, wif.service_account_id)
|
||||
return all(ids) and _has_identity_source()
|
||||
|
||||
|
||||
def is_anthropic_model(model_name: str) -> bool:
|
||||
"""Whether ``model_name`` is routed to Anthropic's own API (``anthropic/…``).
|
||||
|
||||
Only that prefix reaches the LiteLLM route that carries the federated token:
|
||||
a bare ``claude-…`` name goes to the OpenAI route. Claude on Bedrock or
|
||||
Vertex carries that provider's prefix and keeps that provider's auth.
|
||||
"""
|
||||
name = model_name.strip().lower()
|
||||
for prefix in ("litellm/", "any-llm/"):
|
||||
name = name.removeprefix(prefix)
|
||||
return name.startswith("anthropic/")
|
||||
|
||||
|
||||
def applies(model_name: str | None, api_key: str | None = None) -> bool:
|
||||
"""Whether ``model_name`` authenticates by Workload Identity Federation.
|
||||
|
||||
An explicit API key wins, as it does in the Anthropic SDKs.
|
||||
"""
|
||||
if api_key or not model_name:
|
||||
return False
|
||||
return is_anthropic_model(model_name) and is_configured()
|
||||
|
||||
|
||||
# Identity token sources, in the order Strix tries them.
|
||||
SOURCE_FILE = "ANTHROPIC_IDENTITY_TOKEN_FILE"
|
||||
SOURCE_COMMAND = "ANTHROPIC_IDENTITY_TOKEN_COMMAND"
|
||||
SOURCE_LITERAL = "ANTHROPIC_IDENTITY_TOKEN"
|
||||
SOURCE_ENTRA = "Entra ID sign-in (strix auth login entra)"
|
||||
SOURCE_GITHUB = "GitHub Actions OIDC"
|
||||
|
||||
|
||||
def identity_source() -> str | None:
|
||||
"""The source the next exchange reads its JWT from, or None if there is none."""
|
||||
wif = load_settings().anthropic_wif
|
||||
if wif.identity_token_file:
|
||||
return SOURCE_FILE
|
||||
if (wif.identity_token_command or "").strip():
|
||||
return SOURCE_COMMAND
|
||||
if (os.environ.get("ANTHROPIC_IDENTITY_TOKEN") or "").strip():
|
||||
return SOURCE_LITERAL
|
||||
if entra_user.is_configured():
|
||||
return SOURCE_ENTRA
|
||||
if _github_oidc_available():
|
||||
return SOURCE_GITHUB
|
||||
return None
|
||||
|
||||
|
||||
def _has_identity_source() -> bool:
|
||||
return identity_source() is not None
|
||||
|
||||
|
||||
def _github_oidc_available() -> bool:
|
||||
return bool(
|
||||
os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL")
|
||||
and os.environ.get("ACTIONS_ID_TOKEN_REQUEST_TOKEN")
|
||||
)
|
||||
|
||||
|
||||
def read_identity_token() -> str:
|
||||
"""The IdP JWT for one exchange, read fresh every time.
|
||||
|
||||
JWTs carrying a ``jti`` are single-use, and projected token files rotate on
|
||||
disk, so a cached JWT would be rejected on the next exchange.
|
||||
"""
|
||||
wif = load_settings().anthropic_wif
|
||||
source = identity_source()
|
||||
token_file = wif.identity_token_file
|
||||
if source == SOURCE_FILE and token_file:
|
||||
try:
|
||||
token = Path(token_file).expanduser().read_text(encoding="utf-8").strip()
|
||||
except OSError as exc:
|
||||
raise WifAuthError(f"cannot read ANTHROPIC_IDENTITY_TOKEN_FILE: {exc}") from exc
|
||||
if not token:
|
||||
raise WifAuthError(f"ANTHROPIC_IDENTITY_TOKEN_FILE {token_file} is empty")
|
||||
return token
|
||||
if source == SOURCE_COMMAND:
|
||||
return _run_token_command((wif.identity_token_command or "").strip())
|
||||
if source == SOURCE_LITERAL:
|
||||
return (os.environ.get("ANTHROPIC_IDENTITY_TOKEN") or "").strip()
|
||||
if source == SOURCE_ENTRA:
|
||||
try:
|
||||
return entra_user.get_valid_token()
|
||||
except entra_user.EntraAuthError as exc:
|
||||
raise WifAuthError(
|
||||
f"Entra ID sign-in failed ({exc}); run: strix auth login entra"
|
||||
) from exc
|
||||
if source == SOURCE_GITHUB:
|
||||
return _fetch_github_oidc_token()
|
||||
raise WifAuthError(
|
||||
"no identity token: set ANTHROPIC_IDENTITY_TOKEN_FILE, "
|
||||
"ANTHROPIC_IDENTITY_TOKEN_COMMAND or ANTHROPIC_IDENTITY_TOKEN, "
|
||||
"or run: strix auth login entra"
|
||||
)
|
||||
|
||||
|
||||
def _run_token_command(command: str) -> str:
|
||||
"""Run ``ANTHROPIC_IDENTITY_TOKEN_COMMAND`` and return its stdout as the JWT.
|
||||
|
||||
The command is split like a shell would but run without one, so pipelines
|
||||
need an explicit ``sh -c '...'``. Errors never include stdout, which may
|
||||
hold a token.
|
||||
"""
|
||||
try:
|
||||
args = shlex.split(command)
|
||||
except ValueError as exc:
|
||||
raise WifAuthError(f"cannot parse ANTHROPIC_IDENTITY_TOKEN_COMMAND: {exc}") from exc
|
||||
if not args:
|
||||
raise WifAuthError("ANTHROPIC_IDENTITY_TOKEN_COMMAND is empty")
|
||||
if args[0].startswith("~"):
|
||||
# Only then: Path() would also turn "./cmd" into a PATH lookup of "cmd".
|
||||
args[0] = str(Path(args[0]).expanduser())
|
||||
try:
|
||||
result = subprocess.run( # noqa: S603
|
||||
args,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
stdin=subprocess.DEVNULL,
|
||||
timeout=_TIMEOUT_S,
|
||||
check=False,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise WifAuthError(f"ANTHROPIC_IDENTITY_TOKEN_COMMAND could not run: {exc}") from exc
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
raise WifAuthError(
|
||||
f"ANTHROPIC_IDENTITY_TOKEN_COMMAND timed out after {_TIMEOUT_S}s"
|
||||
) from exc
|
||||
if result.returncode != 0:
|
||||
stderr = result.stderr.strip().splitlines()
|
||||
detail = f": {stderr[-1][:200]}" if stderr else ""
|
||||
raise WifAuthError(
|
||||
f"ANTHROPIC_IDENTITY_TOKEN_COMMAND exited with status {result.returncode}{detail}"
|
||||
)
|
||||
token = result.stdout.strip()
|
||||
if not token:
|
||||
raise WifAuthError("ANTHROPIC_IDENTITY_TOKEN_COMMAND printed no token")
|
||||
if any(char.isspace() for char in token):
|
||||
raise WifAuthError("ANTHROPIC_IDENTITY_TOKEN_COMMAND must print a single JWT")
|
||||
return token
|
||||
|
||||
|
||||
def _fetch_github_oidc_token() -> str:
|
||||
url = os.environ["ACTIONS_ID_TOKEN_REQUEST_URL"]
|
||||
try:
|
||||
resp = requests.get(
|
||||
url,
|
||||
params={"audience": GITHUB_AUDIENCE},
|
||||
headers={"Authorization": f"Bearer {os.environ['ACTIONS_ID_TOKEN_REQUEST_TOKEN']}"},
|
||||
timeout=_TIMEOUT_S,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
raise WifAuthError(f"GitHub OIDC token request failed: {exc}") from exc
|
||||
if resp.status_code != 200:
|
||||
raise WifAuthError(f"GitHub OIDC token request failed: HTTP {resp.status_code}")
|
||||
value = _json(resp).get("value")
|
||||
if not isinstance(value, str) or not value:
|
||||
raise WifAuthError("GitHub OIDC token response has no value")
|
||||
return value
|
||||
|
||||
|
||||
def exchange(assertion: str | None = None) -> tuple[str, float]:
|
||||
"""Run one ``jwt-bearer`` exchange; return ``(access_token, expires_at)``.
|
||||
|
||||
``assertion`` is the IdP JWT to present; by default it is read from the
|
||||
configured source.
|
||||
"""
|
||||
settings = load_settings()
|
||||
wif = settings.anthropic_wif
|
||||
body: dict[str, str] = {
|
||||
"grant_type": GRANT_TYPE,
|
||||
"assertion": assertion or read_identity_token(),
|
||||
"federation_rule_id": wif.federation_rule_id or "",
|
||||
"organization_id": wif.organization_id or "",
|
||||
"service_account_id": wif.service_account_id or "",
|
||||
}
|
||||
if wif.workspace_id:
|
||||
body["workspace_id"] = wif.workspace_id
|
||||
url = token_url(settings.llm.api_base)
|
||||
started = time.time()
|
||||
try:
|
||||
resp = requests.post(url, json=body, timeout=_TIMEOUT_S)
|
||||
except requests.RequestException as exc:
|
||||
raise WifAuthError(f"token exchange request failed: {exc}") from exc
|
||||
payload = _json(resp)
|
||||
if resp.status_code != 200:
|
||||
raise WifAuthError(_describe_error(resp, payload))
|
||||
access = payload.get("access_token")
|
||||
expires_in = payload.get("expires_in")
|
||||
if not isinstance(access, str) or not access or not isinstance(expires_in, int | float):
|
||||
raise WifAuthError("token exchange response is missing access_token or expires_in")
|
||||
logger.info("Anthropic workload identity token minted (expires in %ss)", int(expires_in))
|
||||
return access, started + float(expires_in)
|
||||
|
||||
|
||||
def token_url(api_base: str | None) -> str:
|
||||
"""The exchange endpoint on the same host the Messages calls go to."""
|
||||
base = (api_base or DEFAULT_BASE_URL).rstrip("/")
|
||||
base = base.removesuffix("/v1")
|
||||
return f"{base}{TOKEN_PATH}"
|
||||
|
||||
|
||||
def _json(resp: requests.Response) -> dict[str, Any]:
|
||||
try:
|
||||
data: object = resp.json()
|
||||
except ValueError:
|
||||
return {}
|
||||
return cast("dict[str, Any]", data) if isinstance(data, dict) else {}
|
||||
|
||||
|
||||
def _describe_error(resp: requests.Response, payload: dict[str, Any]) -> str:
|
||||
error = payload.get("error")
|
||||
detail = ""
|
||||
if isinstance(error, dict):
|
||||
fields = cast("dict[str, Any]", error)
|
||||
detail = f"{fields.get('type', '')}: {fields.get('message', '')}".strip(": ")
|
||||
elif isinstance(error, str):
|
||||
detail = f"{error}: {payload.get('error_description', '')}".strip(": ")
|
||||
request_id = resp.headers.get("request-id")
|
||||
message = f"token exchange failed: HTTP {resp.status_code}"
|
||||
if detail:
|
||||
message += f" ({detail})"
|
||||
if request_id:
|
||||
message += f" [request-id {request_id}]"
|
||||
if resp.status_code == 401:
|
||||
message += "; see the authentication history under Settings > Workload identity"
|
||||
return message
|
||||
|
||||
|
||||
def get_valid_token() -> str:
|
||||
"""A usable access token, exchanging a new one when the cached one runs low.
|
||||
|
||||
Inside the advisory window a failed exchange keeps serving the cached token;
|
||||
inside the mandatory window it raises.
|
||||
"""
|
||||
global _cached # noqa: PLW0603
|
||||
with _lock:
|
||||
now = time.time()
|
||||
if _cached is not None and now < _cached[1] - _ADVISORY_REFRESH_S:
|
||||
return _cached[0]
|
||||
try:
|
||||
_cached = exchange()
|
||||
except WifAuthError:
|
||||
# The failed exchange (token command plus HTTP) can take a minute.
|
||||
if _cached is not None and time.time() < _cached[1] - _MANDATORY_REFRESH_S:
|
||||
logger.warning(
|
||||
"Anthropic workload identity refresh failed; using the cached token",
|
||||
exc_info=True,
|
||||
)
|
||||
return _cached[0]
|
||||
raise
|
||||
return _cached[0]
|
||||
|
||||
|
||||
def reset_cache() -> None:
|
||||
global _cached # noqa: PLW0603
|
||||
with _lock:
|
||||
_cached = None
|
||||
|
|
@ -403,4 +403,11 @@ def subscription_model(model_name: str | None) -> str | None:
|
|||
|
||||
|
||||
def auth_mode(model_name: str | None) -> str:
|
||||
return "subscription" if subscription_model(model_name) else "api_key"
|
||||
if subscription_model(model_name):
|
||||
return "subscription"
|
||||
from strix.config import anthropic_wif
|
||||
from strix.config.loader import load_settings
|
||||
|
||||
if anthropic_wif.applies(model_name, load_settings().llm.api_key):
|
||||
return "workload_identity"
|
||||
return "api_key"
|
||||
|
|
|
|||
304
strix/config/entra_user.py
Normal file
304
strix/config/entra_user.py
Normal file
|
|
@ -0,0 +1,304 @@
|
|||
"""Microsoft Entra ID user sign-in, as an identity source for Anthropic WIF.
|
||||
|
||||
``strix auth login entra`` signs the user in with OAuth 2.0 + PKCE against their
|
||||
own Entra tenant, as an app registration that is both the OAuth client and the
|
||||
token audience (``api://<app>/user_impersonation``). The refresh token is kept in
|
||||
``~/.strix/entra-auth.json``; Anthropic Workload Identity Federation then
|
||||
exchanges the user's Entra access token for a short-lived Anthropic token.
|
||||
|
||||
Because the client and the audience are the same app, the stored refresh token
|
||||
can only mint tokens for that audience -- not Microsoft Graph or Azure
|
||||
management, as a refresh token issued to the Azure CLI could.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import contextlib
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
import urllib.parse
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
|
||||
import requests
|
||||
|
||||
from strix.config.loader import load_settings
|
||||
from strix.utils.secret_files import write_secret_text
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Generator
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
PROVIDER = "entra"
|
||||
|
||||
LOGIN_BASE_URL = "https://login.microsoftonline.com"
|
||||
CALLBACK_HOST = "localhost"
|
||||
CALLBACK_PORT = 8400
|
||||
# Entra matches loopback redirects on scheme, host and path (any port), so this must
|
||||
# stay the bare ``http://localhost`` registered as the app's public-client redirect.
|
||||
CALLBACK_PATH = "/"
|
||||
REDIRECT_URI = f"http://{CALLBACK_HOST}:{CALLBACK_PORT}"
|
||||
|
||||
_TOKEN_TIMEOUT = 30
|
||||
_EXPIRY_SKEW_S = 300
|
||||
|
||||
_refresh_lock = threading.Lock()
|
||||
|
||||
# Kept separate from cli-config.json so OAuth tokens never land in the env-var config.
|
||||
AUTH_PATH = Path.home() / ".strix" / "entra-auth.json"
|
||||
|
||||
|
||||
class EntraAuthError(Exception):
|
||||
def __init__(self, code: str, message: str | None = None) -> None:
|
||||
self.code = code
|
||||
super().__init__(message or code)
|
||||
|
||||
|
||||
def scope_for(app_id: str) -> str:
|
||||
return f"api://{app_id}/user_impersonation"
|
||||
|
||||
|
||||
def configured_ids() -> tuple[str | None, str | None]:
|
||||
"""``(tenant, app)`` from ``STRIX_ENTRA_TENANT_ID`` / ``STRIX_ENTRA_APP_ID``."""
|
||||
wif = load_settings().anthropic_wif
|
||||
return (wif.entra_tenant_id or "").strip() or None, (wif.entra_app_id or "").strip() or None
|
||||
|
||||
|
||||
def read_record() -> dict[str, Any] | None:
|
||||
try:
|
||||
data = json.loads(AUTH_PATH.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
record = cast("dict[str, Any]", data)
|
||||
if record.get("provider") != PROVIDER:
|
||||
return None
|
||||
if not all(record.get(key) for key in ("tenant", "app", "access", "refresh")):
|
||||
return None
|
||||
return record
|
||||
|
||||
|
||||
def save_record(record: dict[str, Any]) -> None:
|
||||
"""Store a new sign-in, after any refresh in flight (this or another process)."""
|
||||
with _refresh_guard():
|
||||
_write_record(record)
|
||||
|
||||
|
||||
def _write_record(record: dict[str, Any]) -> None:
|
||||
write_secret_text(AUTH_PATH, json.dumps(record, indent=2))
|
||||
|
||||
|
||||
def is_authenticated() -> bool:
|
||||
return read_record() is not None
|
||||
|
||||
|
||||
def logout() -> bool:
|
||||
"""Delete the stored sign-in; return whether there was one.
|
||||
|
||||
Waits for any refresh in flight, so the refresh can't recreate the file.
|
||||
"""
|
||||
with _refresh_guard():
|
||||
existed = AUTH_PATH.exists()
|
||||
with contextlib.suppress(OSError):
|
||||
AUTH_PATH.unlink()
|
||||
return existed
|
||||
|
||||
|
||||
def is_configured() -> bool:
|
||||
"""Signed in, for the tenant and app the settings name (when they name one)."""
|
||||
record = read_record()
|
||||
if record is None:
|
||||
return False
|
||||
tenant, app = configured_ids()
|
||||
return (tenant is None or tenant == record["tenant"]) and (app is None or app == record["app"])
|
||||
|
||||
|
||||
def build_authorize_url(tenant: str, app: str, challenge: str, state: str) -> str:
|
||||
params = {
|
||||
"client_id": app,
|
||||
"response_type": "code",
|
||||
# Query (not form_post), so a pasted redirect URL carries the code.
|
||||
"response_mode": "query",
|
||||
"redirect_uri": REDIRECT_URI,
|
||||
"scope": f"{scope_for(app)} offline_access openid profile",
|
||||
"code_challenge": challenge,
|
||||
"code_challenge_method": "S256",
|
||||
"state": state,
|
||||
"prompt": "select_account",
|
||||
}
|
||||
return f"{_tenant_url(tenant)}/oauth2/v2.0/authorize?{urllib.parse.urlencode(params)}"
|
||||
|
||||
|
||||
def exchange_code(tenant: str, app: str, code: str, verifier: str) -> dict[str, Any]:
|
||||
data = _post_form(
|
||||
tenant,
|
||||
{
|
||||
"grant_type": "authorization_code",
|
||||
"client_id": app,
|
||||
"code": code,
|
||||
"code_verifier": verifier,
|
||||
"redirect_uri": REDIRECT_URI,
|
||||
"scope": f"{scope_for(app)} offline_access",
|
||||
},
|
||||
)
|
||||
return _record_from_token_response(data, tenant, app)
|
||||
|
||||
|
||||
def refresh_tokens(record: dict[str, Any]) -> dict[str, Any]:
|
||||
data = _post_form(
|
||||
record["tenant"],
|
||||
{
|
||||
"grant_type": "refresh_token",
|
||||
"client_id": record["app"],
|
||||
"refresh_token": record["refresh"],
|
||||
"scope": f"{scope_for(record['app'])} offline_access",
|
||||
},
|
||||
)
|
||||
return _record_from_token_response(
|
||||
data,
|
||||
record["tenant"],
|
||||
record["app"],
|
||||
refresh_fallback=record["refresh"],
|
||||
username=record.get("username"),
|
||||
)
|
||||
|
||||
|
||||
def get_valid_token() -> str:
|
||||
"""A usable Entra access token, refreshed under a cross-process guard when near expiry."""
|
||||
record = _require_record()
|
||||
if not _near_expiry(record):
|
||||
return str(record["access"])
|
||||
with _refresh_guard():
|
||||
record = _require_record()
|
||||
if not _near_expiry(record):
|
||||
return str(record["access"])
|
||||
try:
|
||||
refreshed = refresh_tokens(record)
|
||||
except EntraAuthError:
|
||||
# A peer process may have refreshed (and rotated the refresh token) first.
|
||||
latest = read_record()
|
||||
if latest and latest["refresh"] != record["refresh"] and not _near_expiry(latest):
|
||||
return str(latest["access"])
|
||||
raise
|
||||
_write_record(refreshed)
|
||||
return str(refreshed["access"])
|
||||
|
||||
|
||||
def jwt_claims(token: str | None) -> dict[str, Any]:
|
||||
"""Unverified JWT payload, for display only (Entra and Anthropic verify on use)."""
|
||||
if not token or token.count(".") != 2:
|
||||
return {}
|
||||
payload = token.split(".")[1]
|
||||
try:
|
||||
data = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
|
||||
except ValueError:
|
||||
return {}
|
||||
return cast("dict[str, Any]", data) if isinstance(data, dict) else {}
|
||||
|
||||
|
||||
def _tenant_url(tenant: str) -> str:
|
||||
return f"{LOGIN_BASE_URL}/{urllib.parse.quote(tenant, safe='')}"
|
||||
|
||||
|
||||
def _require_record() -> dict[str, Any]:
|
||||
record = read_record()
|
||||
if record is None:
|
||||
raise EntraAuthError("not_authenticated", "not signed in; run: strix auth login entra")
|
||||
return record
|
||||
|
||||
|
||||
def _near_expiry(record: dict[str, Any]) -> bool:
|
||||
expires_at = record.get("expires_at")
|
||||
if not isinstance(expires_at, int | float):
|
||||
return True
|
||||
return expires_at - _EXPIRY_SKEW_S <= time.time()
|
||||
|
||||
|
||||
def _post_form(tenant: str, payload: dict[str, str]) -> dict[str, Any]:
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{_tenant_url(tenant)}/oauth2/v2.0/token",
|
||||
data=payload,
|
||||
headers={"Accept": "application/json"},
|
||||
timeout=_TOKEN_TIMEOUT,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
raise EntraAuthError("unavailable", str(exc)) from exc
|
||||
try:
|
||||
data: object = response.json()
|
||||
except ValueError:
|
||||
data = {}
|
||||
body = cast("dict[str, Any]", data) if isinstance(data, dict) else {}
|
||||
if response.status_code >= 400:
|
||||
raise EntraAuthError(str(body.get("error") or "token_http_error"), _describe_error(body))
|
||||
return body
|
||||
|
||||
|
||||
def _describe_error(body: dict[str, Any]) -> str:
|
||||
"""Entra's ``error: AADSTSxxxx: description``, without the trace and correlation lines."""
|
||||
error = str(body.get("error") or "token request failed")
|
||||
description = str(body.get("error_description") or "").strip().splitlines()
|
||||
return f"{error}: {description[0][:300]}" if description else error
|
||||
|
||||
|
||||
def _record_from_token_response(
|
||||
data: dict[str, Any],
|
||||
tenant: str,
|
||||
app: str,
|
||||
*,
|
||||
refresh_fallback: str | None = None,
|
||||
username: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
access = data.get("access_token")
|
||||
# Entra rotates refresh tokens; keep the old one if a response omits it.
|
||||
refresh = data.get("refresh_token") or refresh_fallback
|
||||
if not isinstance(access, str) or not access:
|
||||
raise EntraAuthError("bad_response", "token response missing access_token")
|
||||
if not isinstance(refresh, str) or not refresh:
|
||||
raise EntraAuthError("bad_response", "token response missing refresh_token")
|
||||
id_claims = jwt_claims(data.get("id_token") if isinstance(data.get("id_token"), str) else None)
|
||||
expires_in = data.get("expires_in")
|
||||
ttl = expires_in if isinstance(expires_in, int | float) else 3600
|
||||
return {
|
||||
"type": "oauth",
|
||||
"provider": PROVIDER,
|
||||
"tenant": tenant,
|
||||
"app": app,
|
||||
"access": access,
|
||||
"refresh": refresh,
|
||||
"username": id_claims.get("preferred_username") or username,
|
||||
"expires_at": time.time() + float(ttl),
|
||||
}
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _refresh_guard() -> Generator[None]:
|
||||
"""Serialize refreshes, sign-ins and sign-outs within (lock) and across (flock)
|
||||
Strix processes. A refresh holds it from reading the record to saving the new
|
||||
one, so a login or logout can't land in between and be overwritten."""
|
||||
with _refresh_lock:
|
||||
try:
|
||||
import fcntl
|
||||
|
||||
lock_path = AUTH_PATH.with_suffix(".lock")
|
||||
lock_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
handle = lock_path.open("w")
|
||||
except (ImportError, OSError):
|
||||
yield
|
||||
return
|
||||
try:
|
||||
with contextlib.suppress(OSError):
|
||||
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
|
||||
yield
|
||||
finally:
|
||||
with contextlib.suppress(OSError):
|
||||
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
||||
handle.close()
|
||||
|
|
@ -4,6 +4,7 @@ from __future__ import annotations
|
|||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import functools
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
|
|
@ -39,7 +40,7 @@ from openai.types.responses.response_usage import ResponseUsage
|
|||
from openai.types.shared import Reasoning
|
||||
|
||||
from strix.agents.prompt import CACHE_POINT
|
||||
from strix.config import codex
|
||||
from strix.config import anthropic_wif, codex
|
||||
from strix.config.loader import load_settings
|
||||
from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
|
||||
from strix.config.tool_call_limits import TurnToolCallLimiter
|
||||
|
|
@ -50,6 +51,7 @@ if TYPE_CHECKING:
|
|||
from collections.abc import AsyncIterator
|
||||
|
||||
from agents.agent_output import AgentOutputSchemaBase
|
||||
from agents.extensions.models.litellm_model import LitellmModel
|
||||
from agents.handoffs import Handoff
|
||||
from agents.items import ModelResponse, TResponseInputItem, TResponseStreamEvent
|
||||
from agents.models.interface import ModelTracing
|
||||
|
|
@ -549,6 +551,39 @@ class _CredentialedLitellmProvider(ModelProvider):
|
|||
)
|
||||
|
||||
|
||||
@functools.cache
|
||||
def _wif_litellm_model_cls() -> type[LitellmModel]:
|
||||
from agents.extensions.models.litellm_model import LitellmModel
|
||||
|
||||
class _WifLitellmModel(LitellmModel):
|
||||
"""An Anthropic model authenticated by Workload Identity Federation.
|
||||
|
||||
Every request re-reads the access token, so a scan outlives each
|
||||
short-lived token; the exchange runs off the event loop.
|
||||
"""
|
||||
|
||||
async def _fetch_response(self, *args: Any, **kwargs: Any) -> Any:
|
||||
self.api_key = await asyncio.to_thread(anthropic_wif.get_valid_token)
|
||||
return await super()._fetch_response(*args, **kwargs)
|
||||
|
||||
return _WifLitellmModel
|
||||
|
||||
|
||||
class _WifLitellmProvider(ModelProvider):
|
||||
"""LiteLLM route that federates Anthropic models and leaves the rest alone."""
|
||||
|
||||
def __init__(self, fallback: ModelProvider) -> None:
|
||||
self._fallback = fallback
|
||||
|
||||
def get_model(self, model_name: str | None) -> Model:
|
||||
from agents.models.default_models import get_default_model
|
||||
|
||||
name = model_name or get_default_model()
|
||||
if not anthropic_wif.is_anthropic_model(name):
|
||||
return self._fallback.get_model(model_name)
|
||||
return _wif_litellm_model_cls()(model=name)
|
||||
|
||||
|
||||
class StrixProvider(MultiProvider):
|
||||
"""Route any non-OpenAI prefix through LiteLLM with the prefix preserved,
|
||||
so users type ``deepseek/deepseek-chat`` rather than
|
||||
|
|
@ -580,7 +615,14 @@ class StrixProvider(MultiProvider):
|
|||
def _create_fallback_provider(self, prefix: str) -> ModelProvider:
|
||||
if prefix == "litellm" and (self._override_api_key or self._override_base_url):
|
||||
return _CredentialedLitellmProvider(self._override_api_key, self._override_base_url)
|
||||
return super()._create_fallback_provider(prefix)
|
||||
provider = super()._create_fallback_provider(prefix)
|
||||
if (
|
||||
prefix == "litellm"
|
||||
and not load_settings().llm.api_key
|
||||
and anthropic_wif.is_configured()
|
||||
):
|
||||
return _WifLitellmProvider(provider)
|
||||
return provider
|
||||
|
||||
def _resolve_prefixed_model(
|
||||
self,
|
||||
|
|
@ -699,6 +741,9 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
|
|||
return
|
||||
_configure_litellm_compatibility()
|
||||
_configure_openrouter_attribution(llm.model)
|
||||
if anthropic_wif.applies(llm.model, llm.api_key):
|
||||
# Fail at startup, not on the first turn, when the federation is misconfigured.
|
||||
anthropic_wif.get_valid_token()
|
||||
if llm.api_key:
|
||||
set_default_openai_key(llm.api_key, use_for_tracing=False)
|
||||
_configure_litellm_default("api_key", llm.api_key)
|
||||
|
|
|
|||
|
|
@ -202,11 +202,36 @@ class ViewerSettings(BaseSettings):
|
|||
app_url: str = Field(default="https://app.strix.ai", alias="STRIX_APP_URL")
|
||||
|
||||
|
||||
class AnthropicWifSettings(BaseSettings):
|
||||
"""Anthropic Workload Identity Federation, under the Anthropic SDKs' own env names.
|
||||
|
||||
``ANTHROPIC_IDENTITY_TOKEN_COMMAND`` is Strix-only: a command whose stdout is
|
||||
the JWT. The literal ``ANTHROPIC_IDENTITY_TOKEN`` JWT is read straight from
|
||||
the environment instead, so it never lands in the persisted config file.
|
||||
"""
|
||||
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
federation_rule_id: str | None = Field(default=None, alias="ANTHROPIC_FEDERATION_RULE_ID")
|
||||
organization_id: str | None = Field(default=None, alias="ANTHROPIC_ORGANIZATION_ID")
|
||||
service_account_id: str | None = Field(default=None, alias="ANTHROPIC_SERVICE_ACCOUNT_ID")
|
||||
workspace_id: str | None = Field(default=None, alias="ANTHROPIC_WORKSPACE_ID")
|
||||
identity_token_file: str | None = Field(default=None, alias="ANTHROPIC_IDENTITY_TOKEN_FILE")
|
||||
identity_token_command: str | None = Field(
|
||||
default=None, alias="ANTHROPIC_IDENTITY_TOKEN_COMMAND"
|
||||
)
|
||||
# Defaults for `strix auth login entra`: the Entra tenant, and the app registration
|
||||
# that is both the OAuth client and the token audience.
|
||||
entra_tenant_id: str | None = Field(default=None, alias="STRIX_ENTRA_TENANT_ID")
|
||||
entra_app_id: str | None = Field(default=None, alias="STRIX_ENTRA_APP_ID")
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
llm: LlmSettings = Field(default_factory=LlmSettings)
|
||||
dedupe: DedupeSettings = Field(default_factory=DedupeSettings)
|
||||
anthropic_wif: AnthropicWifSettings = Field(default_factory=AnthropicWifSettings)
|
||||
runtime: RuntimeSettings = Field(default_factory=RuntimeSettings)
|
||||
context: ContextSettings = Field(default_factory=ContextSettings)
|
||||
telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings)
|
||||
|
|
|
|||
|
|
@ -1,8 +1,11 @@
|
|||
"""`strix auth` — ChatGPT subscription sign-in (login / status / logout).
|
||||
"""`strix auth` — sign-in for ChatGPT subscriptions and Microsoft Entra ID.
|
||||
|
||||
Signing in only stores OAuth tokens (``~/.strix/subscription-auth.json``); model
|
||||
selection stays with ``STRIX_LLM``. A ``chatgpt/<model>`` STRIX_LLM runs on the
|
||||
subscription.
|
||||
``login chatgpt`` stores ChatGPT OAuth tokens (``~/.strix/subscription-auth.json``);
|
||||
a ``chatgpt/<model>`` STRIX_LLM then runs on the subscription. ``login entra``
|
||||
stores an Entra ID user sign-in (``~/.strix/entra-auth.json``) that Anthropic
|
||||
Workload Identity Federation uses as its identity token. ``status`` and
|
||||
``logout`` cover both providers. ``test`` runs one Anthropic federation exchange
|
||||
with the identity token source a scan would use.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -10,7 +13,9 @@ from __future__ import annotations
|
|||
import argparse
|
||||
import base64
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
import time
|
||||
import webbrowser
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
from pathlib import Path
|
||||
|
|
@ -21,7 +26,7 @@ from rich.console import Console
|
|||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
from strix.config import codex, load_settings
|
||||
from strix.config import anthropic_wif, apply_config_override, codex, entra_user, load_settings
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -37,8 +42,16 @@ _CALLBACK_TIMEOUT_S = 300
|
|||
# command and messaging say. ``codex`` is accepted as an alias.
|
||||
LOGIN_PROVIDER = "chatgpt"
|
||||
_ACCEPTED_PROVIDERS = frozenset({LOGIN_PROVIDER, codex.PROVIDER})
|
||||
ENTRA_PROVIDER = entra_user.PROVIDER
|
||||
|
||||
_USAGE = "Usage:\n strix auth login chatgpt [--manual]\n strix auth status\n strix auth logout"
|
||||
_USAGE = (
|
||||
"Usage:\n"
|
||||
" strix auth login chatgpt [--manual]\n"
|
||||
" strix auth login entra [--manual] [--tenant ID] [--app ID]\n"
|
||||
" strix auth status\n"
|
||||
" strix auth logout\n"
|
||||
" strix auth test [--config PATH] test Anthropic Workload Identity Federation"
|
||||
)
|
||||
|
||||
|
||||
def run_auth(argv: list[str]) -> int:
|
||||
|
|
@ -56,6 +69,7 @@ def run_auth(argv: list[str]) -> int:
|
|||
"login": lambda: _login(console, rest),
|
||||
"status": lambda: _status(console),
|
||||
"logout": lambda: _logout(console),
|
||||
"test": lambda: _test(console, rest),
|
||||
}
|
||||
handler = handlers.get(subcommand)
|
||||
if handler is not None:
|
||||
|
|
@ -72,22 +86,32 @@ def _login(console: Console, argv: list[str]) -> int:
|
|||
"provider",
|
||||
nargs="?",
|
||||
default=LOGIN_PROVIDER,
|
||||
help="Model provider to sign in with (default: chatgpt).",
|
||||
help="Provider to sign in with: chatgpt (default) or entra.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--manual",
|
||||
action="store_true",
|
||||
help="Skip the local callback server and paste the redirect URL by hand.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--tenant", help="Entra tenant ID (entra only; default: STRIX_ENTRA_TENANT_ID)."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--app",
|
||||
help="Entra app registration (client) ID (entra only; default: STRIX_ENTRA_APP_ID).",
|
||||
)
|
||||
try:
|
||||
args = parser.parse_args(argv)
|
||||
except SystemExit as exc: # argparse already printed the message
|
||||
return int(exc.code or 2)
|
||||
|
||||
if args.provider.lower() == ENTRA_PROVIDER:
|
||||
return _login_entra(console, args)
|
||||
if args.provider.lower() not in _ACCEPTED_PROVIDERS:
|
||||
console.print(
|
||||
f"[red]Unsupported provider:[/] {args.provider}. "
|
||||
f"Only '{LOGIN_PROVIDER}' (ChatGPT subscription) is supported."
|
||||
f"Use '{LOGIN_PROVIDER}' (ChatGPT subscription) or '{ENTRA_PROVIDER}' "
|
||||
"(Microsoft Entra ID, for Anthropic Workload Identity Federation)."
|
||||
)
|
||||
return 2
|
||||
|
||||
|
|
@ -115,6 +139,60 @@ def _login(console: Console, argv: list[str]) -> int:
|
|||
return 0
|
||||
|
||||
|
||||
def _login_entra(console: Console, args: argparse.Namespace) -> int:
|
||||
default_tenant, default_app = entra_user.configured_ids()
|
||||
tenant = (args.tenant or default_tenant or "").strip()
|
||||
app = (args.app or default_app or "").strip()
|
||||
if not tenant or not app:
|
||||
console.print(
|
||||
"[red]Missing Entra IDs.[/] Pass [cyan]--tenant[/] and [cyan]--app[/], or set "
|
||||
"[cyan]STRIX_ENTRA_TENANT_ID[/] and [cyan]STRIX_ENTRA_APP_ID[/]."
|
||||
)
|
||||
return 2
|
||||
|
||||
verifier, challenge = codex.generate_pkce()
|
||||
state = codex.create_state()
|
||||
authorize_url = entra_user.build_authorize_url(tenant, app, challenge, state)
|
||||
|
||||
console.print()
|
||||
console.print("[bold]Signing in with Microsoft Entra ID[/] [dim](provider: entra)[/]")
|
||||
console.print(
|
||||
"[dim]Anthropic Workload Identity Federation will exchange this sign-in for "
|
||||
"short-lived Anthropic tokens.[/]"
|
||||
)
|
||||
if os.environ.get("SSH_CONNECTION") and not args.manual:
|
||||
console.print(
|
||||
"[yellow]Over SSH:[/] the browser can't reach this host's localhost. Re-run with "
|
||||
f"[cyan]--manual[/], or tunnel port {entra_user.CALLBACK_PORT} "
|
||||
f"([cyan]ssh -L {entra_user.CALLBACK_PORT}:127.0.0.1:{entra_user.CALLBACK_PORT} …[/])."
|
||||
)
|
||||
console.print()
|
||||
|
||||
try:
|
||||
record = _run_oauth_flow(
|
||||
console,
|
||||
authorize_url,
|
||||
verifier,
|
||||
state,
|
||||
manual=args.manual,
|
||||
exchange=lambda code, code_verifier: entra_user.exchange_code(
|
||||
tenant, app, code, code_verifier
|
||||
),
|
||||
port=entra_user.CALLBACK_PORT,
|
||||
path=entra_user.CALLBACK_PATH,
|
||||
account_label="your Microsoft Entra ID account",
|
||||
)
|
||||
except (codex.CodexAuthError, entra_user.EntraAuthError) as exc:
|
||||
return _fail(console, exc)
|
||||
except KeyboardInterrupt:
|
||||
console.print("\n[yellow]Sign-in cancelled.[/]")
|
||||
return 130
|
||||
|
||||
entra_user.save_record(record)
|
||||
_print_entra_success(console, record)
|
||||
return 0
|
||||
|
||||
|
||||
def _run_oauth_flow(
|
||||
console: Console,
|
||||
authorize_url: str,
|
||||
|
|
@ -122,12 +200,20 @@ def _run_oauth_flow(
|
|||
state: str,
|
||||
*,
|
||||
manual: bool,
|
||||
exchange: Callable[[str, str], dict[str, Any]] | None = None,
|
||||
port: int = codex.CALLBACK_PORT,
|
||||
path: str = codex.CALLBACK_PATH,
|
||||
account_label: str = "your ChatGPT subscription",
|
||||
) -> dict[str, Any]:
|
||||
"""Drive the browser (or manual) OAuth flow and return a token record."""
|
||||
server = None if manual else _try_start_callback_server()
|
||||
"""Drive the browser (or manual) OAuth flow and return a token record.
|
||||
|
||||
``exchange(code, verifier)`` redeems the code; it defaults to the ChatGPT one.
|
||||
"""
|
||||
server = None if manual else _try_start_callback_server(port, path, account_label)
|
||||
|
||||
console.print("Open this URL in your browser to authorize:")
|
||||
console.print(f"[cyan]{authorize_url}[/]")
|
||||
# soft_wrap: no hard line breaks inside the URL, so it copies as one piece.
|
||||
console.print(f"[cyan]{authorize_url}[/]", soft_wrap=True)
|
||||
console.print()
|
||||
if not manual:
|
||||
try:
|
||||
|
|
@ -143,7 +229,9 @@ def _run_oauth_flow(
|
|||
code, returned_state, error = result
|
||||
if error:
|
||||
raise codex.CodexAuthError("oauth_error", error)
|
||||
return _finish(code, returned_state, verifier, state, require_state=True)
|
||||
return _finish(
|
||||
code, returned_state, verifier, state, require_state=True, exchange=exchange
|
||||
)
|
||||
console.print("[yellow]Timed out waiting for the browser. Falling back to manual paste.[/]")
|
||||
|
||||
# Manual fallback: the user completes sign-in and pastes the redirect URL
|
||||
|
|
@ -155,7 +243,7 @@ def _run_oauth_flow(
|
|||
except EOFError as exc:
|
||||
raise codex.CodexAuthError("no_input", "no redirect URL provided") from exc
|
||||
code, returned_state = codex.parse_redirect_input(pasted)
|
||||
return _finish(code, returned_state, verifier, state, require_state=False)
|
||||
return _finish(code, returned_state, verifier, state, require_state=False, exchange=exchange)
|
||||
|
||||
|
||||
def _finish(
|
||||
|
|
@ -165,6 +253,7 @@ def _finish(
|
|||
expected_state: str,
|
||||
*,
|
||||
require_state: bool,
|
||||
exchange: Callable[[str, str], dict[str, Any]] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
if not code:
|
||||
raise codex.CodexAuthError("no_code", "no authorization code found in the redirect")
|
||||
|
|
@ -176,7 +265,7 @@ def _finish(
|
|||
raise codex.CodexAuthError("state_mismatch", "missing state in callback; possible CSRF")
|
||||
if returned_state is not None and returned_state != expected_state:
|
||||
raise codex.CodexAuthError("state_mismatch", "state did not match; possible CSRF")
|
||||
return codex.exchange_code(code, verifier)
|
||||
return (exchange or codex.exchange_code)(code, verifier)
|
||||
|
||||
|
||||
class _CallbackServer:
|
||||
|
|
@ -203,7 +292,11 @@ class _CallbackServer:
|
|||
self._httpd.server_close()
|
||||
|
||||
|
||||
def _try_start_callback_server() -> _CallbackServer | None:
|
||||
def _try_start_callback_server(
|
||||
port: int = codex.CALLBACK_PORT,
|
||||
path: str = codex.CALLBACK_PATH,
|
||||
account_label: str = "your ChatGPT subscription",
|
||||
) -> _CallbackServer | None:
|
||||
event = threading.Event()
|
||||
holder: dict[str, Any] = {}
|
||||
|
||||
|
|
@ -213,15 +306,17 @@ def _try_start_callback_server() -> _CallbackServer | None:
|
|||
|
||||
def do_GET(self) -> None:
|
||||
parsed = urlparse(self.path)
|
||||
if parsed.path != codex.CALLBACK_PATH:
|
||||
query = parse_qs(parsed.query)
|
||||
# With a bare "/" callback path, ignore stray requests (favicon, probes)
|
||||
# that carry neither a code nor an error.
|
||||
if parsed.path != path or not ("code" in query or "error" in query):
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
query = parse_qs(parsed.query)
|
||||
holder["code"] = _first(query, "code")
|
||||
holder["state"] = _first(query, "state")
|
||||
holder["error"] = _first(query, "error_description") or _first(query, "error")
|
||||
body = _render_callback_html().encode("utf-8")
|
||||
body = _render_callback_html(account_label).encode("utf-8")
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
|
|
@ -230,9 +325,9 @@ def _try_start_callback_server() -> _CallbackServer | None:
|
|||
event.set()
|
||||
|
||||
try:
|
||||
httpd = HTTPServer(("127.0.0.1", codex.CALLBACK_PORT), Handler)
|
||||
httpd = HTTPServer(("127.0.0.1", port), Handler)
|
||||
except OSError:
|
||||
logger.debug("could not bind callback port %d", codex.CALLBACK_PORT, exc_info=True)
|
||||
logger.debug("could not bind callback port %d", port, exc_info=True)
|
||||
return None
|
||||
return _CallbackServer(httpd, event, holder)
|
||||
|
||||
|
|
@ -243,12 +338,23 @@ def _first(query: dict[str, list[str]], key: str) -> str | None:
|
|||
|
||||
|
||||
def _status(console: Console) -> int:
|
||||
"""Report every provider; exit 0 if any of them is signed in."""
|
||||
chatgpt = _status_chatgpt(console)
|
||||
console.print()
|
||||
entra = _status_entra(console)
|
||||
return 0 if chatgpt or entra else 1
|
||||
|
||||
|
||||
def _status_chatgpt(console: Console) -> bool:
|
||||
console.print("[bold]ChatGPT subscription[/]")
|
||||
record = codex.read_record()
|
||||
if record is None:
|
||||
console.print("[yellow]Not signed in.[/] Run [cyan]strix auth login chatgpt[/] to sign in.")
|
||||
return 1
|
||||
console.print(
|
||||
" [yellow]Not signed in.[/] Run [cyan]strix auth login chatgpt[/] to sign in."
|
||||
)
|
||||
return False
|
||||
settings = load_settings()
|
||||
console.print("[green]Signed in[/] with a ChatGPT subscription.")
|
||||
console.print(" [green]Signed in[/] with a ChatGPT subscription.")
|
||||
console.print(f" Account: [bold]{record.get('account_id')}[/]")
|
||||
if codex.subscription_model(settings.llm.model):
|
||||
console.print(f" Runs use the subscription (STRIX_LLM=[bold]{settings.llm.model}[/]).")
|
||||
|
|
@ -257,16 +363,128 @@ def _status(console: Console) -> int:
|
|||
" [yellow]Note:[/] set [cyan]STRIX_LLM[/] to e.g. [cyan]chatgpt/gpt-5.4[/] "
|
||||
"to run on the subscription."
|
||||
)
|
||||
return 0
|
||||
return True
|
||||
|
||||
|
||||
def _status_entra(console: Console) -> bool:
|
||||
console.print("[bold]Microsoft Entra ID[/] [dim](Anthropic Workload Identity Federation)[/]")
|
||||
record = entra_user.read_record()
|
||||
if record is None:
|
||||
console.print(" [yellow]Not signed in.[/] Run [cyan]strix auth login entra[/] to sign in.")
|
||||
return False
|
||||
console.print(f" [green]Signed in[/] as [bold]{record.get('username') or 'unknown user'}[/]")
|
||||
console.print(f" Tenant: {record['tenant']}", soft_wrap=True)
|
||||
console.print(f" App: {record['app']}", soft_wrap=True)
|
||||
expires_at = record.get("expires_at")
|
||||
left = int(expires_at - time.time()) if isinstance(expires_at, int | float) else 0
|
||||
if left > 0:
|
||||
console.print(f" Access token: valid for {left // 60}m (refreshed automatically)")
|
||||
else:
|
||||
console.print(" Access token: expired (refreshed on next use)")
|
||||
claims = entra_user.jwt_claims(record.get("access"))
|
||||
aud = claims.get("aud")
|
||||
console.print(f" aud: {'OK' if aud == record['app'] else f'[red]MISMATCH ({aud})[/]'}")
|
||||
console.print(f" azp: {claims.get('azp')}", soft_wrap=True)
|
||||
console.print(f" oid: {claims.get('oid')}", soft_wrap=True)
|
||||
console.print(f" sub: {claims.get('sub')}", soft_wrap=True)
|
||||
if not entra_user.is_configured():
|
||||
console.print(
|
||||
" [yellow]Not used:[/] STRIX_ENTRA_TENANT_ID / STRIX_ENTRA_APP_ID name a different "
|
||||
"tenant or app. Sign in again or update them."
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def _logout(console: Console) -> int:
|
||||
"""Sign out of every provider and say which ones had a stored sign-in."""
|
||||
cleared: list[str] = []
|
||||
if codex.read_record() is not None:
|
||||
cleared.append("ChatGPT subscription")
|
||||
codex.logout()
|
||||
console.print("[green]Signed out.[/] Stored subscription credentials removed.")
|
||||
if entra_user.logout():
|
||||
cleared.append("Microsoft Entra ID")
|
||||
if not cleared:
|
||||
console.print("Not signed in to any provider.")
|
||||
return 0
|
||||
console.print(f"[green]Signed out[/] of {' and '.join(cleared)}. Stored credentials removed.")
|
||||
if "Microsoft Entra ID" in cleared:
|
||||
console.print(
|
||||
"[dim]The Entra refresh token was deleted on this host only; revoke the user's "
|
||||
"sessions in Entra to invalidate it everywhere.[/]"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _fail(console: Console, exc: codex.CodexAuthError) -> int:
|
||||
def _test(console: Console, argv: list[str]) -> int:
|
||||
"""Run one Anthropic federation exchange the way a scan would, and report it.
|
||||
|
||||
Prints the identity token's claims and the minted token's prefix, never a
|
||||
token value.
|
||||
"""
|
||||
parser = argparse.ArgumentParser(prog="strix auth test", add_help=True)
|
||||
parser.add_argument("--config", type=Path, help="Strix config file (as for a scan)")
|
||||
args = parser.parse_args(argv)
|
||||
if args.config is not None:
|
||||
apply_config_override(args.config)
|
||||
|
||||
settings = load_settings()
|
||||
wif = settings.anthropic_wif
|
||||
console.print("[bold]Anthropic Workload Identity Federation[/]")
|
||||
missing = [
|
||||
name
|
||||
for name, value in (
|
||||
("ANTHROPIC_FEDERATION_RULE_ID", wif.federation_rule_id),
|
||||
("ANTHROPIC_ORGANIZATION_ID", wif.organization_id),
|
||||
("ANTHROPIC_SERVICE_ACCOUNT_ID", wif.service_account_id),
|
||||
)
|
||||
if not value
|
||||
]
|
||||
if missing:
|
||||
console.print(f" [red]Not configured:[/] set {', '.join(missing)}.", soft_wrap=True)
|
||||
return 1
|
||||
source = anthropic_wif.identity_source()
|
||||
if source is None:
|
||||
console.print(
|
||||
" [red]No identity token source:[/] set ANTHROPIC_IDENTITY_TOKEN_FILE or "
|
||||
"ANTHROPIC_IDENTITY_TOKEN_COMMAND, or run [cyan]strix auth login entra[/]."
|
||||
)
|
||||
return 1
|
||||
console.print(f" Identity token source: [bold]{source}[/]", soft_wrap=True)
|
||||
|
||||
try:
|
||||
assertion = anthropic_wif.read_identity_token()
|
||||
except anthropic_wif.WifAuthError as exc:
|
||||
console.print(f" [red]Identity token failed:[/] {exc}", soft_wrap=True)
|
||||
return 1
|
||||
claims = entra_user.jwt_claims(assertion)
|
||||
for claim in ("iss", "aud", "azp", "oid", "sub"):
|
||||
if claims.get(claim):
|
||||
console.print(f" {claim + ':':<7} {claims[claim]}", soft_wrap=True)
|
||||
|
||||
try:
|
||||
access, expires_at = anthropic_wif.exchange(assertion)
|
||||
except anthropic_wif.WifAuthError as exc:
|
||||
console.print(f" [red]Exchange failed:[/] {exc}", soft_wrap=True)
|
||||
return 1
|
||||
console.print(
|
||||
f" [green]Exchange OK:[/] {access[:13]}… valid for {int(expires_at - time.time())}s"
|
||||
)
|
||||
|
||||
model = settings.llm.model or ""
|
||||
if settings.llm.api_key:
|
||||
console.print(
|
||||
" [yellow]Note:[/] LLM_API_KEY is set, so scans use it instead of federation. "
|
||||
"Set it to empty to use federation."
|
||||
)
|
||||
elif not anthropic_wif.is_anthropic_model(model):
|
||||
console.print(
|
||||
f" [yellow]Note:[/] STRIX_LLM={model or '(unset)'} is not an anthropic/ model, "
|
||||
"so scans don't use federation."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _fail(console: Console, exc: Exception) -> int:
|
||||
error_text = Text()
|
||||
error_text.append("SIGN-IN FAILED", style="bold red")
|
||||
error_text.append("\n\n", style="white")
|
||||
|
|
@ -311,6 +529,45 @@ def _print_success(console: Console) -> None:
|
|||
console.print()
|
||||
|
||||
|
||||
def _print_entra_success(console: Console, record: dict[str, Any]) -> None:
|
||||
claims = entra_user.jwt_claims(record.get("access"))
|
||||
text = Text()
|
||||
text.append(
|
||||
f"Signed in to Microsoft Entra ID as {record.get('username') or 'unknown user'}",
|
||||
style="bold #22c55e",
|
||||
)
|
||||
text.append("\n\n", style="white")
|
||||
text.append("For the Anthropic federation rule: ", style="white")
|
||||
text.append(f"oid {claims.get('oid')}, sub {claims.get('sub')}", style="bold white")
|
||||
text.append("\n\n", style="white")
|
||||
text.append("Set ", style="white")
|
||||
text.append("STRIX_LLM", style="bold white")
|
||||
text.append(" to an ", style="white")
|
||||
text.append("anthropic/", style="bold cyan")
|
||||
text.append(" model and the ", style="white")
|
||||
text.append("ANTHROPIC_FEDERATION_RULE_ID", style="bold white")
|
||||
text.append(", ", style="white")
|
||||
text.append("_ORGANIZATION_ID", style="bold white")
|
||||
text.append(", ", style="white")
|
||||
text.append("_SERVICE_ACCOUNT_ID", style="bold white")
|
||||
text.append(" and ", style="white")
|
||||
text.append("_WORKSPACE_ID", style="bold white")
|
||||
text.append(" variables, with no ", style="white")
|
||||
text.append("LLM_API_KEY", style="bold white")
|
||||
text.append(". Scans then federate with this sign-in.", style="white")
|
||||
console.print()
|
||||
console.print(
|
||||
Panel(
|
||||
text,
|
||||
title="[bold white]STRIX",
|
||||
title_align="left",
|
||||
border_style="#22c55e",
|
||||
padding=(1, 2),
|
||||
)
|
||||
)
|
||||
console.print()
|
||||
|
||||
|
||||
_LOGO_PATH = Path(__file__).resolve().parent.parent / "viewer" / "static" / "logo.png"
|
||||
|
||||
|
||||
|
|
@ -329,8 +586,10 @@ def _logo_img_tag() -> str:
|
|||
return f'<img class="logo" src="data:image/png;base64,{encoded}" alt="" />'
|
||||
|
||||
|
||||
def _render_callback_html() -> str:
|
||||
return _CALLBACK_HTML.replace("<!--LOGO-->", _logo_img_tag())
|
||||
def _render_callback_html(account_label: str = "your ChatGPT subscription") -> str:
|
||||
return _CALLBACK_HTML.replace("<!--LOGO-->", _logo_img_tag()).replace(
|
||||
"<!--ACCOUNT-->", account_label
|
||||
)
|
||||
|
||||
|
||||
_CALLBACK_HTML = """<!doctype html>
|
||||
|
|
@ -399,7 +658,7 @@ _CALLBACK_HTML = """<!doctype html>
|
|||
<h1>You're signed in</h1>
|
||||
<main class="card">
|
||||
<div class="badge">✓</div>
|
||||
<p class="msg">Strix is connected to your ChatGPT subscription. Head back to your
|
||||
<p class="msg">Strix is connected to <!--ACCOUNT-->. Head back to your
|
||||
terminal — your security test runs there.</p>
|
||||
<div class="rule"></div>
|
||||
<p class="tagline">Autonomous AI hackers that <b>find and fix</b> your app's
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@ from rich.console import Console
|
|||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
from strix.config import IntegrationSettings, codex, load_settings
|
||||
from strix.config import IntegrationSettings, anthropic_wif, codex, load_settings
|
||||
from strix.interface.utils import (
|
||||
check_docker_connection,
|
||||
image_exists,
|
||||
|
|
@ -53,7 +53,30 @@ def validate_environment() -> None:
|
|||
if not settings.llm.model:
|
||||
missing_required_vars.append("STRIX_LLM")
|
||||
|
||||
if not settings.llm.api_key:
|
||||
uses_wif = anthropic_wif.applies(settings.llm.model, settings.llm.api_key)
|
||||
if uses_wif:
|
||||
try:
|
||||
anthropic_wif.get_valid_token()
|
||||
except anthropic_wif.WifAuthError as exc:
|
||||
console.print(
|
||||
f"[red]Anthropic workload identity federation failed:[/] {exc}",
|
||||
highlight=False,
|
||||
)
|
||||
report_error("workload_identity_exchange_failed")
|
||||
sys.exit(1)
|
||||
logger.info("Anthropic workload identity token exchange OK")
|
||||
elif (
|
||||
settings.llm.api_key
|
||||
and anthropic_wif.is_anthropic_model(settings.llm.model or "")
|
||||
and anthropic_wif.is_configured()
|
||||
):
|
||||
console.print(
|
||||
"[yellow]LLM_API_KEY is set, so it is used instead of the configured "
|
||||
"Anthropic workload identity federation.[/] Unset it to federate.",
|
||||
highlight=False,
|
||||
)
|
||||
|
||||
if not settings.llm.api_key and not uses_wif:
|
||||
missing_optional_vars.append("LLM_API_KEY")
|
||||
|
||||
if not settings.llm.api_base:
|
||||
|
|
|
|||
476
tests/test_anthropic_wif.py
Normal file
476
tests/test_anthropic_wif.py
Normal file
|
|
@ -0,0 +1,476 @@
|
|||
"""Tests for Anthropic Workload Identity Federation: exchange, refresh, routing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
from agents.extensions.models.litellm_model import LitellmModel
|
||||
|
||||
from strix.config import anthropic_wif, codex, entra_user, loader
|
||||
from strix.config.models import StrixProvider
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
_WIF_ENV = {
|
||||
"ANTHROPIC_FEDERATION_RULE_ID": "fdrl_rule",
|
||||
"ANTHROPIC_ORGANIZATION_ID": "00000000-0000-0000-0000-000000000000",
|
||||
"ANTHROPIC_SERVICE_ACCOUNT_ID": "svac_sa",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def token_file(tmp_path: Path) -> Path:
|
||||
path = tmp_path / "jwt"
|
||||
path.write_text("jwt-1\n", encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolated(monkeypatch: pytest.MonkeyPatch, tmp_path: Path, token_file: Path) -> None:
|
||||
for key in (
|
||||
"STRIX_LLM",
|
||||
"LLM_API_KEY",
|
||||
"OPENAI_API_KEY",
|
||||
"LLM_API_BASE",
|
||||
"OPENAI_API_BASE",
|
||||
"OPENAI_BASE_URL",
|
||||
"LITELLM_BASE_URL",
|
||||
"OLLAMA_API_BASE",
|
||||
"ANTHROPIC_WORKSPACE_ID",
|
||||
"ANTHROPIC_IDENTITY_TOKEN",
|
||||
"ANTHROPIC_IDENTITY_TOKEN_COMMAND",
|
||||
"ACTIONS_ID_TOKEN_REQUEST_URL",
|
||||
"ACTIONS_ID_TOKEN_REQUEST_TOKEN",
|
||||
"STRIX_ENTRA_TENANT_ID",
|
||||
"STRIX_ENTRA_APP_ID",
|
||||
):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
for key, value in _WIF_ENV.items():
|
||||
monkeypatch.setenv(key, value)
|
||||
monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN_FILE", str(token_file))
|
||||
monkeypatch.setattr(entra_user, "AUTH_PATH", tmp_path / "entra-auth.json")
|
||||
monkeypatch.setattr(loader, "_override", tmp_path / "missing-config.json")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
anthropic_wif.reset_cache()
|
||||
|
||||
|
||||
def _response(status: int, body: dict[str, Any], headers: dict[str, str] | None = None) -> Any:
|
||||
resp = mock.MagicMock(spec=requests.Response)
|
||||
resp.status_code = status
|
||||
resp.json.return_value = body
|
||||
resp.headers = headers or {}
|
||||
return resp
|
||||
|
||||
|
||||
def _ok(token: str = "sk-ant-oat01-a", expires_in: int = 600) -> Any: # noqa: S107
|
||||
return _response(200, {"access_token": token, "token_type": "Bearer", "expires_in": expires_in})
|
||||
|
||||
|
||||
def test_configured_only_with_ids_and_a_token_source(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
assert anthropic_wif.is_configured()
|
||||
monkeypatch.delenv("ANTHROPIC_IDENTITY_TOKEN_FILE")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
assert not anthropic_wif.is_configured()
|
||||
monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN", "jwt-literal")
|
||||
assert anthropic_wif.is_configured()
|
||||
monkeypatch.delenv("ANTHROPIC_SERVICE_ACCOUNT_ID")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
assert not anthropic_wif.is_configured()
|
||||
|
||||
|
||||
def test_exchange_sends_jwt_bearer_grant(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("ANTHROPIC_WORKSPACE_ID", "wrkspc_ws")
|
||||
with mock.patch.object(anthropic_wif.requests, "post", return_value=_ok()) as post:
|
||||
token, expires_at = anthropic_wif.exchange()
|
||||
assert token == "sk-ant-oat01-a" # noqa: S105
|
||||
assert expires_at > 0
|
||||
url = post.call_args.args[0]
|
||||
assert url == "https://api.anthropic.com/v1/oauth/token"
|
||||
assert post.call_args.kwargs["json"] == {
|
||||
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
|
||||
"assertion": "jwt-1",
|
||||
"federation_rule_id": "fdrl_rule",
|
||||
"organization_id": "00000000-0000-0000-0000-000000000000",
|
||||
"service_account_id": "svac_sa",
|
||||
"workspace_id": "wrkspc_ws",
|
||||
}
|
||||
|
||||
|
||||
def test_exchange_omits_unset_workspace() -> None:
|
||||
with mock.patch.object(anthropic_wif.requests, "post", return_value=_ok()) as post:
|
||||
anthropic_wif.exchange()
|
||||
assert "workspace_id" not in post.call_args.kwargs["json"]
|
||||
|
||||
|
||||
def test_token_url_follows_api_base() -> None:
|
||||
assert anthropic_wif.token_url("https://gw.example/v1/") == "https://gw.example/v1/oauth/token"
|
||||
assert anthropic_wif.token_url(None) == "https://api.anthropic.com/v1/oauth/token"
|
||||
|
||||
|
||||
def test_exchange_rereads_the_token_file(token_file: Path) -> None:
|
||||
with mock.patch.object(anthropic_wif.requests, "post", return_value=_ok()) as post:
|
||||
anthropic_wif.exchange()
|
||||
token_file.write_text("jwt-2", encoding="utf-8")
|
||||
anthropic_wif.exchange()
|
||||
assertions = [call.kwargs["json"]["assertion"] for call in post.call_args_list]
|
||||
assert assertions == ["jwt-1", "jwt-2"]
|
||||
|
||||
|
||||
def test_exchange_error_names_status_but_not_the_jwt() -> None:
|
||||
body = {
|
||||
"type": "error",
|
||||
"error": {"type": "authentication_error", "message": "Authentication failed"},
|
||||
}
|
||||
resp = _response(401, body, {"request-id": "req_1"})
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.requests, "post", return_value=resp),
|
||||
pytest.raises(anthropic_wif.WifAuthError) as exc_info,
|
||||
):
|
||||
anthropic_wif.exchange()
|
||||
message = str(exc_info.value)
|
||||
assert "HTTP 401" in message
|
||||
assert "authentication_error" in message
|
||||
assert "req_1" in message
|
||||
assert "jwt-1" not in message
|
||||
|
||||
|
||||
def test_github_actions_oidc_token_used_without_a_file(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("ANTHROPIC_IDENTITY_TOKEN_FILE")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
monkeypatch.setenv("ACTIONS_ID_TOKEN_REQUEST_URL", "https://gh.example/token?x=1")
|
||||
monkeypatch.setenv("ACTIONS_ID_TOKEN_REQUEST_TOKEN", "gh-req")
|
||||
gh = _response(200, {"value": "gh-jwt"})
|
||||
with mock.patch.object(anthropic_wif.requests, "get", return_value=gh) as get:
|
||||
assert anthropic_wif.read_identity_token() == "gh-jwt"
|
||||
assert get.call_args.kwargs["params"] == {"audience": "https://api.anthropic.com"}
|
||||
assert get.call_args.kwargs["headers"] == {"Authorization": "Bearer gh-req"}
|
||||
|
||||
|
||||
def _python_command(code: str) -> str:
|
||||
return shlex.join([sys.executable, "-c", code])
|
||||
|
||||
|
||||
def _use_token_command(monkeypatch: pytest.MonkeyPatch, command: str) -> None:
|
||||
monkeypatch.delenv("ANTHROPIC_IDENTITY_TOKEN_FILE")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN_COMMAND", command)
|
||||
|
||||
|
||||
def _completed(returncode: int = 0, stdout: str = "", stderr: str = "") -> Any:
|
||||
return subprocess.CompletedProcess(["cmd"], returncode, stdout, stderr)
|
||||
|
||||
|
||||
def test_token_command_used_without_a_file(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_use_token_command(monkeypatch, _python_command("print('cmd-jwt')"))
|
||||
assert anthropic_wif.is_configured()
|
||||
assert anthropic_wif.read_identity_token() == "cmd-jwt"
|
||||
|
||||
|
||||
def test_token_file_wins_over_command(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN_COMMAND", "should-not-run")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
with mock.patch.object(anthropic_wif.subprocess, "run") as run:
|
||||
assert anthropic_wif.read_identity_token() == "jwt-1"
|
||||
run.assert_not_called()
|
||||
|
||||
|
||||
def test_token_command_runs_on_every_exchange(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_use_token_command(monkeypatch, "issue-token --audience x")
|
||||
outputs = [_completed(stdout="jwt-a\n"), _completed(stdout="jwt-b\n")]
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.subprocess, "run", side_effect=outputs) as run,
|
||||
mock.patch.object(anthropic_wif.requests, "post", return_value=_ok()) as post,
|
||||
):
|
||||
anthropic_wif.exchange()
|
||||
anthropic_wif.exchange()
|
||||
assert run.call_args.args[0] == ["issue-token", "--audience", "x"]
|
||||
assert run.call_args.kwargs["stdin"] == subprocess.DEVNULL
|
||||
assertions = [call.kwargs["json"]["assertion"] for call in post.call_args_list]
|
||||
assert assertions == ["jwt-a", "jwt-b"]
|
||||
|
||||
|
||||
def test_token_command_failure_names_status_and_stderr_but_not_stdout(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
code = "import sys; print('secret-jwt'); print('boom', file=sys.stderr); sys.exit(3)"
|
||||
_use_token_command(monkeypatch, _python_command(code))
|
||||
with pytest.raises(anthropic_wif.WifAuthError) as exc:
|
||||
anthropic_wif.read_identity_token()
|
||||
assert "status 3" in str(exc.value)
|
||||
assert "boom" in str(exc.value)
|
||||
assert "secret-jwt" not in str(exc.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("stdout", "message"),
|
||||
[("\n", "printed no token"), ("jwt-a jwt-b\n", "single JWT")],
|
||||
)
|
||||
def test_token_command_output_must_be_one_jwt(
|
||||
monkeypatch: pytest.MonkeyPatch, stdout: str, message: str
|
||||
) -> None:
|
||||
_use_token_command(monkeypatch, "issue-token")
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.subprocess, "run", return_value=_completed(stdout=stdout)),
|
||||
pytest.raises(anthropic_wif.WifAuthError, match=message),
|
||||
):
|
||||
anthropic_wif.read_identity_token()
|
||||
|
||||
|
||||
def test_token_command_not_found(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
_use_token_command(monkeypatch, str(tmp_path / "no-such-command"))
|
||||
with pytest.raises(anthropic_wif.WifAuthError, match="could not run"):
|
||||
anthropic_wif.read_identity_token()
|
||||
|
||||
|
||||
def test_token_command_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_use_token_command(monkeypatch, "issue-token")
|
||||
timeout = subprocess.TimeoutExpired(["issue-token"], 30)
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.subprocess, "run", side_effect=timeout),
|
||||
pytest.raises(anthropic_wif.WifAuthError, match="timed out"),
|
||||
):
|
||||
anthropic_wif.read_identity_token()
|
||||
|
||||
|
||||
def test_token_command_expands_home_in_program(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_use_token_command(monkeypatch, "~/bin/issue-token token")
|
||||
monkeypatch.setenv("HOME", "/home/someone")
|
||||
with mock.patch.object(
|
||||
anthropic_wif.subprocess, "run", return_value=_completed(stdout="jwt\n")
|
||||
) as run:
|
||||
anthropic_wif.read_identity_token()
|
||||
assert run.call_args.args[0] == ["/home/someone/bin/issue-token", "token"]
|
||||
|
||||
|
||||
def test_token_command_persists_and_is_read_from_config(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = tmp_path / "config.json"
|
||||
config.write_text('{"env": {}}', encoding="utf-8")
|
||||
monkeypatch.setattr(loader, "_override", config)
|
||||
command = "issue-token --from-config"
|
||||
_use_token_command(monkeypatch, command)
|
||||
loader.persist_current()
|
||||
saved = json.loads(config.read_text(encoding="utf-8"))["env"]
|
||||
assert saved.get("ANTHROPIC_IDENTITY_TOKEN_COMMAND") == command
|
||||
|
||||
monkeypatch.delenv("ANTHROPIC_IDENTITY_TOKEN_COMMAND")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
with mock.patch.object(
|
||||
anthropic_wif.subprocess, "run", return_value=_completed(stdout="jwt\n")
|
||||
) as run:
|
||||
assert anthropic_wif.read_identity_token() == "jwt"
|
||||
assert run.call_args.args[0] == ["issue-token", "--from-config"]
|
||||
|
||||
|
||||
def test_empty_env_clears_saved_token_command(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
config = tmp_path / "config.json"
|
||||
config.write_text(
|
||||
json.dumps({"env": {"ANTHROPIC_IDENTITY_TOKEN_COMMAND": "issue-token"}}), encoding="utf-8"
|
||||
)
|
||||
monkeypatch.setattr(loader, "_override", config)
|
||||
_use_token_command(monkeypatch, "")
|
||||
assert not anthropic_wif.is_configured()
|
||||
loader.persist_current()
|
||||
saved = json.loads(config.read_text(encoding="utf-8"))["env"]
|
||||
assert "ANTHROPIC_IDENTITY_TOKEN_COMMAND" not in saved
|
||||
|
||||
|
||||
def _sign_in_to_entra(**overrides: Any) -> None:
|
||||
record = {
|
||||
"type": "oauth",
|
||||
"provider": "entra",
|
||||
"tenant": "tenant-id",
|
||||
"app": "app-id",
|
||||
"access": "entra-jwt",
|
||||
"refresh": "entra-refresh",
|
||||
"expires_at": time.time() + 3600,
|
||||
}
|
||||
record.update(overrides)
|
||||
entra_user.save_record(record)
|
||||
|
||||
|
||||
def _without_token_file(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("ANTHROPIC_IDENTITY_TOKEN_FILE")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
|
||||
|
||||
def test_entra_sign_in_used_without_other_sources(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_without_token_file(monkeypatch)
|
||||
assert not anthropic_wif.is_configured()
|
||||
_sign_in_to_entra()
|
||||
assert anthropic_wif.is_configured()
|
||||
assert anthropic_wif.read_identity_token() == "entra-jwt"
|
||||
|
||||
|
||||
def test_token_command_wins_over_entra_sign_in(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_sign_in_to_entra()
|
||||
_use_token_command(monkeypatch, "issue-token")
|
||||
with mock.patch.object(
|
||||
anthropic_wif.subprocess, "run", return_value=_completed(stdout="cmd-jwt\n")
|
||||
):
|
||||
assert anthropic_wif.read_identity_token() == "cmd-jwt"
|
||||
|
||||
|
||||
def test_entra_refresh_failure_names_the_fix(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
_without_token_file(monkeypatch)
|
||||
_sign_in_to_entra(expires_at=time.time() - 1)
|
||||
refused = entra_user.EntraAuthError("invalid_grant", "invalid_grant: AADSTS70000")
|
||||
with (
|
||||
mock.patch.object(entra_user, "refresh_tokens", side_effect=refused),
|
||||
pytest.raises(anthropic_wif.WifAuthError, match="strix auth login entra"),
|
||||
):
|
||||
anthropic_wif.read_identity_token()
|
||||
|
||||
|
||||
def test_cached_token_reused_until_advisory_window() -> None:
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.time, "time", return_value=1000.0),
|
||||
mock.patch.object(anthropic_wif.requests, "post", return_value=_ok(expires_in=600)) as post,
|
||||
):
|
||||
assert anthropic_wif.get_valid_token() == "sk-ant-oat01-a"
|
||||
assert anthropic_wif.get_valid_token() == "sk-ant-oat01-a"
|
||||
assert post.call_count == 1
|
||||
|
||||
|
||||
def test_refreshes_inside_advisory_window() -> None:
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.requests, "post", return_value=_ok("sk-ant-oat01-a")),
|
||||
mock.patch.object(anthropic_wif.time, "time", return_value=1000.0),
|
||||
):
|
||||
anthropic_wif.get_valid_token()
|
||||
# expires at 1600; 1500 is inside the 120s advisory window.
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.time, "time", return_value=1500.0),
|
||||
mock.patch.object(anthropic_wif.requests, "post", return_value=_ok("sk-ant-oat01-b")),
|
||||
):
|
||||
assert anthropic_wif.get_valid_token() == "sk-ant-oat01-b"
|
||||
|
||||
|
||||
def test_advisory_refresh_failure_keeps_cached_token() -> None:
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.time, "time", return_value=1000.0),
|
||||
mock.patch.object(anthropic_wif.requests, "post", return_value=_ok()),
|
||||
):
|
||||
anthropic_wif.get_valid_token()
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.time, "time", return_value=1500.0),
|
||||
mock.patch.object(
|
||||
anthropic_wif.requests, "post", side_effect=requests.ConnectionError("down")
|
||||
),
|
||||
):
|
||||
assert anthropic_wif.get_valid_token() == "sk-ant-oat01-a"
|
||||
|
||||
|
||||
def test_mandatory_refresh_failure_raises() -> None:
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.time, "time", return_value=1000.0),
|
||||
mock.patch.object(anthropic_wif.requests, "post", return_value=_ok()),
|
||||
):
|
||||
anthropic_wif.get_valid_token()
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.time, "time", return_value=1580.0),
|
||||
mock.patch.object(
|
||||
anthropic_wif.requests, "post", side_effect=requests.ConnectionError("down")
|
||||
),
|
||||
pytest.raises(anthropic_wif.WifAuthError),
|
||||
):
|
||||
anthropic_wif.get_valid_token()
|
||||
|
||||
|
||||
def test_slow_failed_refresh_does_not_serve_an_expired_token() -> None:
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.time, "time", return_value=1000.0),
|
||||
mock.patch.object(anthropic_wif.requests, "post", return_value=_ok()),
|
||||
):
|
||||
anthropic_wif.get_valid_token()
|
||||
# Expires at 1600. The refresh starts at 1555, inside the advisory window but
|
||||
# outside the mandatory one, and fails only after the token has expired.
|
||||
clock = iter([1555.0, 1555.0, 1610.0, 1610.0])
|
||||
|
||||
def _slow_failure(*_args: Any, **_kwargs: Any) -> Any:
|
||||
raise requests.ConnectionError("down")
|
||||
|
||||
with (
|
||||
mock.patch.object(anthropic_wif.time, "time", side_effect=lambda: next(clock)),
|
||||
mock.patch.object(anthropic_wif.requests, "post", side_effect=_slow_failure),
|
||||
pytest.raises(anthropic_wif.WifAuthError),
|
||||
):
|
||||
anthropic_wif.get_valid_token()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("model", "expected"),
|
||||
[
|
||||
("anthropic/claude-sonnet-5-5", True),
|
||||
("litellm/anthropic/claude-sonnet-5-5", True),
|
||||
("claude-sonnet-5-5", False),
|
||||
("bedrock/anthropic.claude-sonnet-5-5", False),
|
||||
("vertex_ai/claude-sonnet-5-5", False),
|
||||
("openrouter/anthropic/claude-sonnet-5-5", False),
|
||||
("openai/gpt-5", False),
|
||||
],
|
||||
)
|
||||
def test_applies_only_to_anthropic_direct(model: str, *, expected: bool) -> None:
|
||||
assert anthropic_wif.applies(model) is expected
|
||||
|
||||
|
||||
def test_explicit_api_key_wins() -> None:
|
||||
assert not anthropic_wif.applies("anthropic/claude-sonnet-5-5", "sk-ant-api03-x")
|
||||
|
||||
|
||||
def test_auth_mode_reports_workload_identity(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
assert codex.auth_mode("anthropic/claude-sonnet-5-5") == "workload_identity"
|
||||
assert codex.auth_mode("openai/gpt-5") == "api_key"
|
||||
monkeypatch.setenv("LLM_API_KEY", "sk-ant-api03-x")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
assert codex.auth_mode("anthropic/claude-sonnet-5-5") == "api_key"
|
||||
|
||||
|
||||
def _unwrap(model: object) -> object:
|
||||
while hasattr(model, "_inner"):
|
||||
model = model._inner
|
||||
return model
|
||||
|
||||
|
||||
def test_provider_federates_anthropic_requests() -> None:
|
||||
model = _unwrap(StrixProvider().get_model("anthropic/claude-sonnet-5-5"))
|
||||
sent: dict[str, Any] = {}
|
||||
|
||||
async def fake_fetch(self: Any, *_args: Any, **_kwargs: Any) -> str:
|
||||
sent["api_key"] = self.api_key
|
||||
return "response"
|
||||
|
||||
with (
|
||||
mock.patch.object(LitellmModel, "_fetch_response", fake_fetch),
|
||||
mock.patch.object(anthropic_wif, "get_valid_token", return_value="sk-ant-oat01-z"),
|
||||
):
|
||||
result = asyncio.run(model._fetch_response()) # type: ignore[attr-defined]
|
||||
assert result == "response"
|
||||
assert sent["api_key"] == "sk-ant-oat01-z"
|
||||
|
||||
|
||||
def test_provider_leaves_other_litellm_models_alone() -> None:
|
||||
model = _unwrap(StrixProvider().get_model("deepseek/deepseek-chat"))
|
||||
assert type(model).__name__ == "LitellmModel"
|
||||
|
||||
|
||||
def test_provider_uses_api_key_when_set(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("LLM_API_KEY", "sk-ant-api03-x")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
model = _unwrap(StrixProvider().get_model("anthropic/claude-sonnet-5-5"))
|
||||
assert type(model).__name__ == "LitellmModel"
|
||||
|
|
@ -2,11 +2,14 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import time
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.config import codex
|
||||
from strix.config import anthropic_wif, codex, entra_user, loader
|
||||
from strix.interface import auth_cli
|
||||
|
||||
|
||||
|
|
@ -17,6 +20,11 @@ if TYPE_CHECKING:
|
|||
@pytest.fixture(autouse=True)
|
||||
def _tmp_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(codex, "AUTH_PATH", tmp_path / "home" / ".strix" / "subscription-auth.json")
|
||||
monkeypatch.setattr(entra_user, "AUTH_PATH", tmp_path / "home" / ".strix" / "entra-auth.json")
|
||||
for key in ("STRIX_ENTRA_TENANT_ID", "STRIX_ENTRA_APP_ID"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
monkeypatch.setattr(loader, "_override", tmp_path / "missing-config.json")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
|
||||
|
||||
def test_login_provider_is_chatgpt() -> None:
|
||||
|
|
@ -104,3 +112,193 @@ def test_login_accepts_provider_aliases(provider: str, monkeypatch: pytest.Monke
|
|||
|
||||
assert auth_cli.run_auth(["login", provider]) == 0
|
||||
assert reached["flow"] is True
|
||||
|
||||
|
||||
def _entra_record(**overrides: Any) -> dict[str, Any]:
|
||||
record = {
|
||||
"type": "oauth",
|
||||
"provider": "entra",
|
||||
"tenant": "tenant-id",
|
||||
"app": "app-id",
|
||||
"access": "entra-access",
|
||||
"refresh": "entra-refresh",
|
||||
"username": "user@example.com",
|
||||
"expires_at": time.time() + 3600,
|
||||
}
|
||||
record.update(overrides)
|
||||
return record
|
||||
|
||||
|
||||
def _codex_record() -> dict[str, Any]:
|
||||
return {
|
||||
"type": "oauth",
|
||||
"provider": "codex",
|
||||
"access": "a",
|
||||
"refresh": "r",
|
||||
"account_id": "acct",
|
||||
"expires_at": time.time() + 3600,
|
||||
}
|
||||
|
||||
|
||||
def test_login_entra_manual_paste_saves_the_sign_in(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
calls: list[tuple[str, ...]] = []
|
||||
|
||||
def _exchange(tenant: str, app: str, code: str, verifier: str) -> dict[str, Any]:
|
||||
calls.append((tenant, app, code))
|
||||
assert verifier
|
||||
return _entra_record(tenant=tenant, app=app)
|
||||
|
||||
monkeypatch.setattr(entra_user, "exchange_code", _exchange)
|
||||
monkeypatch.setattr(
|
||||
auth_cli.Console, "input", lambda *_args, **_kwargs: "http://localhost:8400/?code=abc"
|
||||
)
|
||||
|
||||
argv = ["login", "entra", "--manual", "--tenant", "tenant-id", "--app", "app-id"]
|
||||
assert auth_cli.run_auth(argv) == 0
|
||||
assert calls == [("tenant-id", "app-id", "abc")]
|
||||
saved = entra_user.read_record()
|
||||
assert saved is not None
|
||||
assert saved["tenant"] == "tenant-id"
|
||||
|
||||
|
||||
def test_login_entra_uses_settings_for_ids(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("STRIX_ENTRA_TENANT_ID", "tenant-from-env")
|
||||
monkeypatch.setenv("STRIX_ENTRA_APP_ID", "app-from-env")
|
||||
seen: dict[str, str] = {}
|
||||
|
||||
def _fake_flow(_console: Any, url: str, *_args: Any, **_kwargs: Any) -> dict[str, Any]:
|
||||
seen["url"] = url
|
||||
return _entra_record()
|
||||
|
||||
monkeypatch.setattr(auth_cli, "_run_oauth_flow", _fake_flow)
|
||||
assert auth_cli.run_auth(["login", "entra"]) == 0
|
||||
assert "/tenant-from-env/oauth2/v2.0/authorize" in seen["url"]
|
||||
assert "client_id=app-from-env" in seen["url"]
|
||||
|
||||
|
||||
def test_login_entra_without_ids_is_a_usage_error(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
def _should_not_run(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
|
||||
msg = "OAuth flow must not start without Entra IDs"
|
||||
raise AssertionError(msg)
|
||||
|
||||
monkeypatch.setattr(auth_cli, "_run_oauth_flow", _should_not_run)
|
||||
assert auth_cli.run_auth(["login", "entra"]) == 2
|
||||
|
||||
|
||||
def test_status_reports_both_providers(capsys: pytest.CaptureFixture[str]) -> None:
|
||||
entra_user.save_record(_entra_record())
|
||||
assert auth_cli.run_auth(["status"]) == 0
|
||||
out = capsys.readouterr().out
|
||||
assert "ChatGPT subscription" in out
|
||||
assert "Not signed in" in out
|
||||
assert "Microsoft Entra ID" in out
|
||||
assert "user@example.com" in out
|
||||
|
||||
|
||||
def test_logout_signs_out_of_every_provider(capsys: pytest.CaptureFixture[str]) -> None:
|
||||
codex.save_record(_codex_record())
|
||||
entra_user.save_record(_entra_record())
|
||||
assert auth_cli.run_auth(["logout"]) == 0
|
||||
out = capsys.readouterr().out
|
||||
assert "ChatGPT subscription and Microsoft Entra ID" in out
|
||||
assert codex.read_record() is None
|
||||
assert entra_user.read_record() is None
|
||||
|
||||
|
||||
def test_logout_when_signed_out(capsys: pytest.CaptureFixture[str]) -> None:
|
||||
assert auth_cli.run_auth(["logout"]) == 0
|
||||
assert "Not signed in" in capsys.readouterr().out
|
||||
|
||||
|
||||
_WIF_ENV = {
|
||||
"ANTHROPIC_FEDERATION_RULE_ID": "fdrl_test",
|
||||
"ANTHROPIC_ORGANIZATION_ID": "org-test",
|
||||
"ANTHROPIC_SERVICE_ACCOUNT_ID": "svac_test",
|
||||
"STRIX_LLM": "anthropic/claude-sonnet-4-6",
|
||||
}
|
||||
|
||||
|
||||
def _fake_jwt(claims: dict[str, Any]) -> str:
|
||||
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=")
|
||||
return f"e30.{payload}.sig"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def wif_env(monkeypatch: pytest.MonkeyPatch) -> str:
|
||||
for key in (
|
||||
"ANTHROPIC_WORKSPACE_ID",
|
||||
"ANTHROPIC_IDENTITY_TOKEN_FILE",
|
||||
"ANTHROPIC_IDENTITY_TOKEN_COMMAND",
|
||||
"ACTIONS_ID_TOKEN_REQUEST_URL",
|
||||
):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
for key, value in _WIF_ENV.items():
|
||||
monkeypatch.setenv(key, value)
|
||||
monkeypatch.setenv("LLM_API_KEY", "")
|
||||
jwt = _fake_jwt({"iss": "https://idp.example", "sub": "subject-123"})
|
||||
monkeypatch.setenv("ANTHROPIC_IDENTITY_TOKEN", jwt)
|
||||
return jwt
|
||||
|
||||
|
||||
def test_test_reports_source_claims_and_prefix_only(
|
||||
wif_env: str, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
seen: list[str | None] = []
|
||||
|
||||
def _exchange(assertion: str | None = None) -> tuple[str, float]:
|
||||
seen.append(assertion)
|
||||
return "sk-ant-oat01-SECRETPART", time.time() + 600
|
||||
|
||||
monkeypatch.setattr(anthropic_wif, "exchange", _exchange)
|
||||
assert auth_cli.run_auth(["test"]) == 0
|
||||
out = capsys.readouterr().out
|
||||
assert seen == [wif_env]
|
||||
assert "ANTHROPIC_IDENTITY_TOKEN" in out
|
||||
assert "subject-123" in out
|
||||
assert "sk-ant-oat01-…" in out
|
||||
assert "SECRETPART" not in out
|
||||
assert wif_env not in out
|
||||
assert "Note:" not in out
|
||||
|
||||
|
||||
def test_test_without_federation_ids(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
for key in _WIF_ENV:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
assert auth_cli.run_auth(["test"]) == 1
|
||||
|
||||
|
||||
def test_test_reports_exchange_failure(
|
||||
wif_env: str, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
del wif_env
|
||||
|
||||
def _fail(_assertion: str | None = None) -> tuple[str, float]:
|
||||
raise anthropic_wif.WifAuthError("token exchange failed: HTTP 401")
|
||||
|
||||
monkeypatch.setattr(anthropic_wif, "exchange", _fail)
|
||||
assert auth_cli.run_auth(["test"]) == 1
|
||||
assert "HTTP 401" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_test_warns_when_an_api_key_wins(
|
||||
wif_env: str, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
del wif_env
|
||||
monkeypatch.setenv("LLM_API_KEY", "sk-ant-api-key")
|
||||
monkeypatch.setattr(anthropic_wif, "exchange", lambda _a=None: ("sk-ant-oat01-x", time.time()))
|
||||
assert auth_cli.run_auth(["test"]) == 0
|
||||
assert "LLM_API_KEY is set" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_test_reads_settings_from_config(
|
||||
wif_env: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
del wif_env
|
||||
config = tmp_path / "strix-config.json"
|
||||
config.write_text(json.dumps({"env": _WIF_ENV}), encoding="utf-8")
|
||||
for key in _WIF_ENV:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
monkeypatch.setattr(anthropic_wif, "exchange", lambda _a=None: ("sk-ant-oat01-x", time.time()))
|
||||
assert auth_cli.run_auth(["test"]) == 1
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
assert auth_cli.run_auth(["test", "--config", str(config)]) == 0
|
||||
|
|
|
|||
208
tests/test_entra_user.py
Normal file
208
tests/test_entra_user.py
Normal file
|
|
@ -0,0 +1,208 @@
|
|||
"""Tests for Microsoft Entra ID user sign-in (the `strix auth login entra` token store)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
import urllib.parse
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.config import entra_user, loader
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
TENANT = "11111111-1111-1111-1111-111111111111"
|
||||
APP = "22222222-2222-2222-2222-222222222222"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolated(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
monkeypatch.setattr(entra_user, "AUTH_PATH", tmp_path / ".strix" / "entra-auth.json")
|
||||
for key in ("STRIX_ENTRA_TENANT_ID", "STRIX_ENTRA_APP_ID"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
monkeypatch.setattr(loader, "_override", tmp_path / "missing-config.json")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
|
||||
|
||||
def _jwt(claims: dict[str, Any]) -> str:
|
||||
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).rstrip(b"=").decode()
|
||||
return f"e30.{payload}.sig"
|
||||
|
||||
|
||||
def _response(status: int, body: dict[str, Any]) -> Any:
|
||||
resp = mock.MagicMock()
|
||||
resp.status_code = status
|
||||
resp.json.return_value = body
|
||||
return resp
|
||||
|
||||
|
||||
def _record(**overrides: Any) -> dict[str, Any]:
|
||||
record = {
|
||||
"type": "oauth",
|
||||
"provider": "entra",
|
||||
"tenant": TENANT,
|
||||
"app": APP,
|
||||
"access": "access-1",
|
||||
"refresh": "refresh-1",
|
||||
"username": "user@example.com",
|
||||
"expires_at": time.time() + 3600,
|
||||
}
|
||||
record.update(overrides)
|
||||
return record
|
||||
|
||||
|
||||
def test_authorize_url_requests_the_app_scope_with_pkce() -> None:
|
||||
url = urllib.parse.urlparse(entra_user.build_authorize_url(TENANT, APP, "chal", "st"))
|
||||
params = dict(urllib.parse.parse_qsl(url.query))
|
||||
assert url.netloc == "login.microsoftonline.com"
|
||||
assert url.path == f"/{TENANT}/oauth2/v2.0/authorize"
|
||||
assert params["client_id"] == APP
|
||||
assert params["scope"] == f"api://{APP}/user_impersonation offline_access openid profile"
|
||||
assert params["redirect_uri"] == "http://localhost:8400"
|
||||
assert params["response_mode"] == "query"
|
||||
assert params["code_challenge"] == "chal"
|
||||
assert params["code_challenge_method"] == "S256"
|
||||
assert params["state"] == "st"
|
||||
|
||||
|
||||
def test_exchange_code_builds_a_record() -> None:
|
||||
body = {
|
||||
"access_token": "access-1",
|
||||
"refresh_token": "refresh-1",
|
||||
"expires_in": 3600,
|
||||
"id_token": _jwt({"preferred_username": "user@example.com"}),
|
||||
}
|
||||
with mock.patch.object(entra_user.requests, "post", return_value=_response(200, body)) as post:
|
||||
record = entra_user.exchange_code(TENANT, APP, "the-code", "the-verifier")
|
||||
sent = post.call_args.kwargs["data"]
|
||||
assert post.call_args.args[0] == f"https://login.microsoftonline.com/{TENANT}/oauth2/v2.0/token"
|
||||
assert sent["grant_type"] == "authorization_code"
|
||||
assert sent["client_id"] == APP
|
||||
assert sent["code"] == "the-code"
|
||||
assert sent["code_verifier"] == "the-verifier"
|
||||
assert record["tenant"] == TENANT
|
||||
assert record["app"] == APP
|
||||
assert record["username"] == "user@example.com"
|
||||
assert record["expires_at"] > time.time() + 3000
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("returned", "expected"), [("refresh-2", "refresh-2"), (None, "refresh-1")]
|
||||
)
|
||||
def test_refresh_keeps_or_rotates_the_refresh_token(returned: str | None, expected: str) -> None:
|
||||
body: dict[str, Any] = {"access_token": "access-2", "expires_in": 3600}
|
||||
if returned:
|
||||
body["refresh_token"] = returned
|
||||
with mock.patch.object(entra_user.requests, "post", return_value=_response(200, body)) as post:
|
||||
record = entra_user.refresh_tokens(_record())
|
||||
assert post.call_args.kwargs["data"]["grant_type"] == "refresh_token"
|
||||
assert record["access"] == "access-2"
|
||||
assert record["refresh"] == expected
|
||||
assert record["username"] == "user@example.com"
|
||||
|
||||
|
||||
def test_token_errors_name_the_reason_but_not_tokens_or_trace() -> None:
|
||||
body = {
|
||||
"error": "invalid_grant",
|
||||
"error_description": "AADSTS70000: Provided grant is invalid.\r\nTrace ID: abc\r\n",
|
||||
}
|
||||
with (
|
||||
mock.patch.object(entra_user.requests, "post", return_value=_response(400, body)),
|
||||
pytest.raises(entra_user.EntraAuthError) as exc,
|
||||
):
|
||||
entra_user.refresh_tokens(_record())
|
||||
assert exc.value.code == "invalid_grant"
|
||||
assert "AADSTS70000" in str(exc.value)
|
||||
assert "Trace ID" not in str(exc.value)
|
||||
assert "refresh-1" not in str(exc.value)
|
||||
|
||||
|
||||
def test_valid_token_is_served_from_the_store() -> None:
|
||||
entra_user.save_record(_record())
|
||||
with mock.patch.object(entra_user.requests, "post") as post:
|
||||
assert entra_user.get_valid_token() == "access-1"
|
||||
post.assert_not_called()
|
||||
|
||||
|
||||
def test_token_near_expiry_is_refreshed_and_saved() -> None:
|
||||
entra_user.save_record(_record(expires_at=time.time() + 60))
|
||||
body = {"access_token": "access-2", "refresh_token": "refresh-2", "expires_in": 3600}
|
||||
with mock.patch.object(entra_user.requests, "post", return_value=_response(200, body)):
|
||||
assert entra_user.get_valid_token() == "access-2"
|
||||
saved = entra_user.read_record()
|
||||
assert saved is not None
|
||||
assert saved["refresh"] == "refresh-2"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("action", ["logout", "login"])
|
||||
def test_logout_and_login_wait_for_a_refresh_in_flight(action: str) -> None:
|
||||
entra_user.save_record(_record(expires_at=time.time() + 60))
|
||||
started, release = threading.Event(), threading.Event()
|
||||
body = {"access_token": "access-2", "refresh_token": "refresh-2", "expires_in": 3600}
|
||||
|
||||
def _slow_post(*_args: Any, **_kwargs: Any) -> Any:
|
||||
started.set()
|
||||
assert release.wait(5)
|
||||
return _response(200, body)
|
||||
|
||||
with mock.patch.object(entra_user.requests, "post", side_effect=_slow_post):
|
||||
refresher = threading.Thread(target=entra_user.get_valid_token)
|
||||
refresher.start()
|
||||
assert started.wait(5)
|
||||
if action == "logout":
|
||||
other = threading.Thread(target=entra_user.logout)
|
||||
else:
|
||||
other = threading.Thread(
|
||||
target=entra_user.save_record, args=(_record(access="new-login"),)
|
||||
)
|
||||
other.start()
|
||||
other.join(0.2)
|
||||
assert other.is_alive() # blocked until the refresh has saved
|
||||
release.set()
|
||||
refresher.join(5)
|
||||
other.join(5)
|
||||
|
||||
saved = entra_user.read_record()
|
||||
if action == "logout":
|
||||
assert saved is None
|
||||
else:
|
||||
assert saved is not None
|
||||
assert saved["access"] == "new-login"
|
||||
|
||||
|
||||
def test_not_signed_in_raises() -> None:
|
||||
with pytest.raises(entra_user.EntraAuthError, match="strix auth login entra"):
|
||||
entra_user.get_valid_token()
|
||||
|
||||
|
||||
def test_store_is_private() -> None:
|
||||
entra_user.save_record(_record())
|
||||
assert entra_user.AUTH_PATH.stat().st_mode & 0o777 == 0o600
|
||||
|
||||
|
||||
def test_configured_only_for_the_signed_in_tenant_and_app(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
assert not entra_user.is_configured()
|
||||
entra_user.save_record(_record())
|
||||
assert entra_user.is_configured()
|
||||
monkeypatch.setenv("STRIX_ENTRA_TENANT_ID", TENANT)
|
||||
monkeypatch.setenv("STRIX_ENTRA_APP_ID", APP)
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
assert entra_user.is_configured()
|
||||
monkeypatch.setenv("STRIX_ENTRA_APP_ID", "33333333-3333-3333-3333-333333333333")
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
assert not entra_user.is_configured()
|
||||
|
||||
|
||||
def test_logout_reports_whether_there_was_a_sign_in() -> None:
|
||||
assert entra_user.logout() is False
|
||||
entra_user.save_record(_record())
|
||||
assert entra_user.logout() is True
|
||||
assert entra_user.read_record() is None
|
||||
Loading…
Add table
Reference in a new issue