recipe-maintainers/recipe-maintainer
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

recipe-maintainer: public snapshot (secrets + deployment plans removed, single commit)

Browse Source 
Sanitized single-commit public mirror of recipe-maintainer.
- Removed test-ssh/.testenv (live creds); added test-ssh/.testenv.example placeholders.
- Removed plans/ and planned-updates/ (deployment-planning docs) so no client/
  deployment domains appear in the public repo.
- All other secret stores were already gitignored.
- docs.coopcloud.tech retained as a submodule (public upstream).
This commit is contained in:
autonomic-bot
2026-06-16 20:18:24 +00:00
commit f283a371bb
253 changed files with 15975 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
utils/README.md Normal file
View File

@ -0,0 +1,63 @@
# Utils
Helper scripts for configuring SSO and other integrations across Co-op Cloud recipes.
All scripts use stdlib only (no pip dependencies) and share `authentik_client.py` for Authentik API interactions.
## Getting an Authentik API Token
All SSO setup scripts require an `--authentik-token`. You can create one from your `akadmin` password in several ways:
**Via curl:**
```bash
curl -s -X POST https://<authentik-domain>/api/v3/core/tokens/ \
-H "Content-Type: application/json" \
-u "akadmin:<your-akadmin-password>" \
-d '{"identifier": "sso-setup", "intent": "api", "description": "Token for SSO setup scripts"}'
```
The response JSON contains a `key` field — that's your token.
To retrieve an existing token's key:
```bash
curl -s https://<authentik-domain>/api/v3/core/tokens/sso-setup/view_key/ \
-u "akadmin:<your-akadmin-password>"
```
**Via the Authentik admin UI:**
Go to **Directory > Tokens and App passwords > Create**, set intent to "API Token", then copy the token value.
## Prerequisites
The target abra apps (CryptPad, Immich, etc.) **must already be deployed** before running the SSO setup scripts. The scripts will fail with a clear error if the app is not found. Authentik resources (OAuth2 providers, applications) are created automatically if they don't exist.
## Scripts
### setup_cryptpad_sso.py
Configures Authentik as the OIDC provider for CryptPad SSO. Ensures the OAuth2 provider/application exist in Authentik, updates the CryptPad abra `.env` file, and inserts the client secret as a Docker secret. Requires the CryptPad abra app to already exist.
```bash
python3 utils/setup_cryptpad_sso.py \
--authentik-domain auth.example.com \
--authentik-token <admin-api-token> \
--cryptpad-domain pad.example.com
```
Optional flags: `--client-id`, `--app-slug`, `--test-user`, `--test-pass`, `--test-email`, `--no-test-user`.
### setup_immich_sso.py
Configures Authentik as the OIDC provider for Immich OAuth. Ensures the OAuth2 provider/application exist in Authentik, then configures Immich OAuth settings via its REST API. Requires Immich to already be deployed and reachable.
```bash
python3 utils/setup_immich_sso.py \
--authentik-domain auth.example.com \
--authentik-token <admin-api-token> \
--immich-domain photos.example.com \
--immich-admin-email admin@example.com \
--immich-admin-pass <password>
```

226
utils/authentik_client.py Normal file
View File

@ -0,0 +1,226 @@
"""Shared Authentik API client for SSO integration scripts.
No pip dependencies -- uses urllib.request + json from stdlib.
"""
import json
import urllib.request
import urllib.error
import urllib.parse
class AuthentikClient:
"""Encapsulates all Authentik API interactions for SSO setup."""
def __init__(self, base_url: str, token: str):
self.base_url = base_url.rstrip("/")
self.api_url = f"{self.base_url}/api/v3"
self.token = token
# -- low-level helpers ---------------------------------------------------
def _request(self, method, endpoint, data=None):
"""Make an authenticated API request. Returns parsed JSON."""
url = f"{self.api_url}{endpoint}"
body = json.dumps(data).encode() if data is not None else None
req = urllib.request.Request(url, data=body, method=method)
req.add_header("Authorization", f"Bearer {self.token}")
req.add_header("Content-Type", "application/json")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read()
return json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
raw = e.read().decode(errors="replace")
raise RuntimeError(
f"Authentik API {method} {endpoint} returned {e.code}: {raw}"
) from e
def _get(self, endpoint):
return self._request("GET", endpoint)
def _post(self, endpoint, data):
return self._request("POST", endpoint, data)
def _delete(self, endpoint):
return self._request("DELETE", endpoint)
# -- resolve UUIDs -------------------------------------------------------
def resolve_uuids(self):
"""Fetch PKs for default flows, signing key, and scope mappings.
Returns a dict with keys:
authorization_flow, invalidation_flow, authentication_flow,
signing_key, scope_openid, scope_email, scope_profile
"""
uuids = {}
# Flows
flow_slugs = {
"authorization_flow": "default-provider-authorization-implicit-consent",
"invalidation_flow": "default-provider-invalidation-flow",
"authentication_flow": "default-authentication-flow",
}
for key, slug in flow_slugs.items():
resp = self._get(f"/flows/instances/?slug={slug}")
results = resp.get("results", [])
if not results:
raise RuntimeError(f"Flow '{slug}' not found in Authentik")
uuids[key] = results[0]["pk"]
print(f" {key}: {uuids[key]}")
# Signing key
resp = self._get(
"/crypto/certificatekeypairs/"
"?name=authentik+Self-signed+Certificate"
)
results = resp.get("results", [])
if not results:
raise RuntimeError("Authentik Self-signed Certificate not found")
uuids["signing_key"] = results[0]["pk"]
print(f" signing_key: {uuids['signing_key']}")
# Scope property mappings
scope_managed = {
"scope_openid": "goauthentik.io/providers/oauth2/scope-openid",
"scope_email": "goauthentik.io/providers/oauth2/scope-email",
"scope_profile": "goauthentik.io/providers/oauth2/scope-profile",
}
for key, managed in scope_managed.items():
resp = self._get(f"/propertymappings/all/?managed={managed}")
results = resp.get("results", [])
if not results:
raise RuntimeError(f"Scope mapping '{managed}' not found")
uuids[key] = results[0]["pk"]
print(f" {key}: {uuids[key]}")
return uuids
# -- ensure OAuth2 provider ----------------------------------------------
def ensure_oauth2_provider(self, name, client_id, redirect_uris, uuids):
"""Create or reuse an OAuth2 provider.
redirect_uris: list of {"matching_mode": "strict", "url": "..."}
Returns (provider_pk, client_secret).
"""
resp = self._get(
f"/providers/oauth2/?search={urllib.parse.quote(name)}"
)
for r in resp.get("results", []):
if r["name"] == name:
pk = r["pk"]
print(f" Provider '{name}' already exists (pk: {pk})")
detail = self._get(f"/providers/oauth2/{pk}/")
return pk, detail["client_secret"]
# Create new provider
provider = self._post("/providers/oauth2/", {
"name": name,
"client_type": "confidential",
"client_id": client_id,
"authorization_flow": uuids["authorization_flow"],
"authentication_flow": uuids["authentication_flow"],
"invalidation_flow": uuids["invalidation_flow"],
"redirect_uris": redirect_uris,
"property_mappings": [
uuids["scope_openid"],
uuids["scope_email"],
uuids["scope_profile"],
],
"signing_key": uuids["signing_key"],
"include_claims_in_id_token": True,
})
pk = provider["pk"]
secret = provider["client_secret"]
print(f" Created provider '{name}' (pk: {pk})")
return pk, secret
# -- ensure application --------------------------------------------------
def ensure_application(self, name, slug, provider_pk, launch_url):
"""Create or reuse an application linked to the provider.
Returns the application slug.
"""
resp = self._get(f"/core/applications/?slug={urllib.parse.quote(slug)}")
for r in resp.get("results", []):
if r["slug"] == slug:
print(f" Application '{slug}' already exists")
return slug
self._post("/core/applications/", {
"name": name,
"slug": slug,
"provider": provider_pk,
"meta_launch_url": launch_url,
})
print(f" Created application '{slug}'")
return slug
# -- ensure test user ----------------------------------------------------
def ensure_test_user(self, username, email, password):
"""Create or reuse a user and set their password.
Returns user_pk.
"""
resp = self._get(
f"/core/users/?search={urllib.parse.quote(username)}"
)
user_pk = None
for r in resp.get("results", []):
if r["username"] == username:
user_pk = r["pk"]
print(f" User '{username}' already exists (pk: {user_pk})")
break
if user_pk is None:
user = self._post("/core/users/", {
"username": username,
"name": "Test User",
"email": email,
"is_active": True,
})
user_pk = user["pk"]
print(f" Created user '{username}' (pk: {user_pk})")
# Always set password
self._post(f"/core/users/{user_pk}/set_password/", {
"password": password,
})
print(f" Password set for '{username}'")
return user_pk
# -- ensure app password -------------------------------------------------
def ensure_app_password(self, user_pk, identifier):
"""Create an APP_PASSWORD token (deletes existing one first).
Returns the app password string.
"""
# Delete existing token if present
resp = self._get(
f"/core/tokens/?identifier={urllib.parse.quote(identifier)}"
)
if resp.get("results"):
try:
self._delete(f"/core/tokens/{identifier}/")
print(f" Deleted existing token '{identifier}'")
except RuntimeError:
pass # Already gone
self._post("/core/tokens/", {
"identifier": identifier,
"intent": "app_password",
"user": user_pk,
"description": "App password for OIDC password grant",
"expiring": False,
})
key_resp = self._get(f"/core/tokens/{identifier}/view_key/")
app_password = key_resp["key"]
print(f" APP_PASSWORD created: {app_password[:10]}...")
return app_password

215
utils/setup_cryptpad_sso.py Executable file
View File

@ -0,0 +1,215 @@
#!/usr/bin/env python3
"""Configure Authentik as the OIDC provider for CryptPad SSO.
Requires both Authentik and CryptPad to already be deployed via abra.
Ensures the OAuth2 provider/application exist in Authentik, updates the
CryptPad abra .env file, and inserts the client secret as a Docker secret.
No pip dependencies -- stdlib only.
Usage:
python3 utils/setup_cryptpad_sso.py \
--authentik-domain auth.example.com \
--authentik-token <admin-api-token> \
--cryptpad-domain pad.example.com
"""
import argparse
import os
import re
import subprocess
import sys
# Allow importing authentik_client from the same directory
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from authentik_client import AuthentikClient
def parse_args():
p = argparse.ArgumentParser(
description="Set up Authentik OIDC integration for CryptPad SSO"
)
p.add_argument("--authentik-domain", required=True,
help="Authentik domain (e.g. auth.example.com)")
p.add_argument("--authentik-token", required=True,
help="Authentik admin API token")
p.add_argument("--cryptpad-domain", required=True,
help="CryptPad domain (e.g. pad.example.com)")
p.add_argument("--client-id", default="cryptpad",
help="OAuth2 client ID (default: cryptpad)")
p.add_argument("--app-slug", default="cryptpad",
help="Authentik application slug (default: cryptpad)")
p.add_argument("--test-user", default="testuser",
help="Test user username (default: testuser)")
p.add_argument("--test-pass", default="testpass123",
help="Test user password (default: testpass123)")
p.add_argument("--test-email", default="testuser@test.example.com",
help="Test user email")
p.add_argument("--no-test-user", action="store_true",
help="Skip test user creation")
return p.parse_args()
def find_env_file(cryptpad_domain):
"""Locate the abra .env file for the CryptPad instance.
Searches ~/.abra/servers/*/cryptpad_domain.env
"""
abra_servers = os.path.expanduser("~/.abra/servers")
if not os.path.isdir(abra_servers):
return None
for server in os.listdir(abra_servers):
candidate = os.path.join(abra_servers, server, f"{cryptpad_domain}.env")
if os.path.isfile(candidate):
return candidate
return None
def update_env_file(env_path, oidc_url, client_id):
"""Update CryptPad env file with SSO settings."""
with open(env_path, "r") as f:
content = f.read()
replacements = {
"COMPOSE_FILE": "compose.yml:compose.sso.yml",
"SSO_ENABLED": "true",
"SSO_PROVIDER_NAME": "Authentik",
"SSO_OIDC_URL": oidc_url,
"SSO_CLIENT_ID": client_id,
"SSO_CLIENT_SECRET_VERSION": "v1",
}
for key, value in replacements.items():
pattern = rf"^{re.escape(key)}=.*$"
replacement = f"{key}={value}"
if re.search(pattern, content, re.MULTILINE):
content = re.sub(pattern, replacement, content, flags=re.MULTILINE)
else:
content = content.rstrip("\n") + f"\n{replacement}\n"
print(f" Set {key}={value}")
# Remove old SSO_CLIENT_SECRET env var if present (now a Docker secret)
if re.search(r"^SSO_CLIENT_SECRET=", content, re.MULTILINE):
content = re.sub(
r"^SSO_CLIENT_SECRET=.*\n?", "", content, flags=re.MULTILINE
)
print(" Removed old SSO_CLIENT_SECRET env var (now a Docker secret)")
with open(env_path, "w") as f:
f.write(content)
def insert_docker_secret(cryptpad_domain, client_secret):
"""Insert client secret as a Docker secret via abra."""
cmd = [
"abra", "app", "secret", "insert",
cryptpad_domain, "sso_client_s", "v1", client_secret,
"--chaos", "--no-input",
]
print(f" Running: abra app secret insert {cryptpad_domain} sso_client_s v1 ***")
# abra secret insert needs a TTY wrapper
wrapped = f'script -qefc "{" ".join(cmd)}" /dev/null'
result = subprocess.run(
wrapped, shell=True, capture_output=True, text=True
)
if result.returncode != 0:
# Secret may already exist -- check stderr
output = result.stdout + result.stderr
if "already exists" in output.lower() or "secret already" in output.lower():
print(" Secret already exists, skipping insert")
else:
print(f" WARNING: secret insert returned {result.returncode}")
if output.strip():
print(f" Output: {output.strip()}")
else:
print(" Inserted sso_client_s v1")
def main():
args = parse_args()
ak_url = f"https://{args.authentik_domain}"
cpad_url = f"https://{args.cryptpad_domain}"
client = AuthentikClient(ak_url, args.authentik_token)
# Step 1: Resolve UUIDs
print("=== Resolving Authentik UUIDs ===")
uuids = client.resolve_uuids()
print()
# Step 2: Create OAuth2 provider
print(f"=== Ensuring OAuth2 provider '{args.client_id}' ===")
redirect_uris = [
{"matching_mode": "strict", "url": f"{cpad_url}/ssoauth"},
]
provider_pk, client_secret = client.ensure_oauth2_provider(
name=args.client_id,
client_id=args.client_id,
redirect_uris=redirect_uris,
uuids=uuids,
)
print(f" Client ID: {args.client_id}")
print(f" Client secret: {client_secret}")
print()
# Step 3: Create application
print(f"=== Ensuring application '{args.app_slug}' ===")
client.ensure_application(
name="CryptPad",
slug=args.app_slug,
provider_pk=provider_pk,
launch_url=cpad_url,
)
print()
# Step 4: Optionally create test user
app_password = None
if not args.no_test_user:
print(f"=== Creating test user '{args.test_user}' ===")
user_pk = client.ensure_test_user(
args.test_user, args.test_email, args.test_pass
)
app_password = client.ensure_app_password(
user_pk, f"{args.test_user}-app-password"
)
print()
# Step 5: Update CryptPad env file
print("=== Updating CryptPad env file ===")
oidc_url = f"{ak_url}/application/o/{args.app_slug}"
env_path = find_env_file(args.cryptpad_domain)
if env_path is None:
print(f" ERROR: CryptPad abra app not found for {args.cryptpad_domain}")
print(" The app must exist before running this script.")
print(" Create it first with: abra app new cryptpad ...")
sys.exit(1)
print(f" Found env file: {env_path}")
update_env_file(env_path, oidc_url, args.client_id)
print()
# Step 6: Insert Docker secret
print("=== Inserting Docker secret ===")
insert_docker_secret(args.cryptpad_domain, client_secret)
# Summary
print()
print("=== Setup complete ===")
print()
print(f" Authentik domain: {args.authentik_domain}")
print(f" CryptPad domain: {args.cryptpad_domain}")
print(f" Client ID: {args.client_id}")
print(f" App slug: {args.app_slug}")
print(f" OIDC URL: {oidc_url}")
if app_password:
print(f" Test user: {args.test_user}")
print(f" App password: {app_password[:10]}...")
print()
print("Redeploy CryptPad to activate SSO:")
print(f" abra app deploy {args.cryptpad_domain} --chaos --force --no-input")
if __name__ == "__main__":
main()

253
utils/setup_immich_sso.py Executable file
View File

@ -0,0 +1,253 @@
#!/usr/bin/env python3
"""Configure Authentik as the OIDC provider for Immich OAuth.
Requires both Authentik and Immich to already be deployed.
Ensures the OAuth2 provider/application exist in Authentik, then configures
Immich OAuth settings via its REST API.
No pip dependencies -- stdlib only.
Usage:
python3 utils/setup_immich_sso.py \
--authentik-domain auth.example.com \
--authentik-token <admin-api-token> \
--immich-domain photos.example.com \
--immich-admin-email admin@example.com \
--immich-admin-pass <password>
"""
import argparse
import json
import os
import sys
import urllib.error
import urllib.request
# Allow importing authentik_client from the same directory
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from authentik_client import AuthentikClient
def parse_args():
p = argparse.ArgumentParser(
description="Set up Authentik OIDC integration for Immich OAuth"
)
p.add_argument("--authentik-domain", required=True,
help="Authentik domain (e.g. auth.example.com)")
p.add_argument("--authentik-token", required=True,
help="Authentik admin API token")
p.add_argument("--immich-domain", required=True,
help="Immich domain (e.g. photos.example.com)")
p.add_argument("--immich-admin-email", required=True,
help="Immich admin account email")
p.add_argument("--immich-admin-pass", required=True,
help="Immich admin account password")
p.add_argument("--immich-admin-name", default="Admin",
help="Immich admin display name (default: Admin)")
p.add_argument("--client-id", default="immich",
help="OAuth2 client ID (default: immich)")
p.add_argument("--app-slug", default="immich",
help="Authentik application slug (default: immich)")
p.add_argument("--test-user", default="testuser",
help="Test user username (default: testuser)")
p.add_argument("--test-pass", default="testpass123",
help="Test user password (default: testpass123)")
p.add_argument("--test-email", default="testuser@test.example.com",
help="Test user email")
p.add_argument("--no-test-user", action="store_true",
help="Skip test user creation")
return p.parse_args()
def immich_request(base_url, method, path, data=None, token=None):
"""Make an HTTP request to the Immich API. Returns parsed JSON."""
url = f"{base_url}{path}"
body = json.dumps(data).encode() if data is not None else None
req = urllib.request.Request(url, data=body, method=method)
req.add_header("Content-Type", "application/json")
if token:
req.add_header("Authorization", f"Bearer {token}")
try:
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read()
return json.loads(raw) if raw else {}
except urllib.error.HTTPError as e:
raw = e.read().decode(errors="replace")
raise RuntimeError(
f"Immich API {method} {path} returned {e.code}: {raw}"
) from e
def setup_immich_admin(immich_url, email, password, name):
"""Create admin account (skip if exists) and login. Returns access token."""
# Try creating admin account
print(" Creating Immich admin account ...")
try:
immich_request(immich_url, "POST", "/api/auth/admin-sign-up", {
"email": email,
"password": password,
"name": name,
})
print(" Admin account created")
except RuntimeError as e:
if "admin" in str(e).lower() or "exists" in str(e).lower():
print(" Admin account already exists, continuing")
else:
print(f" Admin signup response: {e}")
print(" Continuing with login attempt...")
# Login
print(" Logging in as Immich admin ...")
login_resp = immich_request(immich_url, "POST", "/api/auth/login", {
"email": email,
"password": password,
})
access_token = login_resp.get("accessToken")
if not access_token:
raise RuntimeError(f"Could not login to Immich: {login_resp}")
print(f" Logged in (token: {access_token[:10]}...)")
return access_token
def configure_immich_oauth(immich_url, access_token, issuer_url, client_id,
client_secret, app_slug):
"""Fetch current Immich system config and merge OAuth settings."""
print(" Fetching current Immich system config ...")
config = immich_request(
immich_url, "GET", "/api/system-config", token=access_token
)
# Merge OAuth settings
oauth = config.get("oauth", {})
oauth.update({
"enabled": True,
"issuerUrl": issuer_url,
"clientId": client_id,
"clientSecret": client_secret,
"scope": "openid email profile",
"autoRegister": True,
"autoLaunch": False,
"buttonText": "Login with Authentik",
"tokenEndpointAuthMethod": "client_secret_post",
"timeout": 30000,
"mobileOverrideEnabled": False,
"mobileRedirectUri": "",
"signingAlgorithm": "RS256",
"storageLabelClaim": "preferred_username",
"storageQuotaClaim": "immich_quota",
"defaultStorageQuota": 0,
"profileSigningAlgorithm": "none",
"roleClaim": "immich_role",
})
config["oauth"] = oauth
print(" Updating OAuth settings ...")
put_resp = immich_request(
immich_url, "PUT", "/api/system-config", data=config,
token=access_token,
)
oauth_enabled = put_resp.get("oauth", {}).get("enabled", False)
if oauth_enabled:
print(" OAuth enabled successfully")
else:
print(f" WARNING: Could not confirm OAuth enabled")
print(f" Response oauth.enabled: {oauth_enabled}")
def main():
args = parse_args()
ak_url = f"https://{args.authentik_domain}"
immich_url = f"https://{args.immich_domain}"
client = AuthentikClient(ak_url, args.authentik_token)
# Step 1: Resolve UUIDs
print("=== Resolving Authentik UUIDs ===")
uuids = client.resolve_uuids()
print()
# Step 2: Create OAuth2 provider
print(f"=== Ensuring OAuth2 provider '{args.client_id}' ===")
redirect_uris = [
{"matching_mode": "strict", "url": f"{immich_url}/auth/login"},
{"matching_mode": "strict", "url": f"{immich_url}/user-settings"},
{"matching_mode": "strict", "url": "app.immich:///oauth-callback"},
]
provider_pk, client_secret = client.ensure_oauth2_provider(
name=args.client_id,
client_id=args.client_id,
redirect_uris=redirect_uris,
uuids=uuids,
)
print(f" Client ID: {args.client_id}")
print(f" Client secret: {client_secret}")
print()
# Step 3: Create application
print(f"=== Ensuring application '{args.app_slug}' ===")
client.ensure_application(
name="Immich",
slug=args.app_slug,
provider_pk=provider_pk,
launch_url=immich_url,
)
print()
# Step 4: Optionally create test user
app_password = None
if not args.no_test_user:
print(f"=== Creating test user '{args.test_user}' ===")
user_pk = client.ensure_test_user(
args.test_user, args.test_email, args.test_pass
)
app_password = client.ensure_app_password(
user_pk, f"{args.test_user}-app-password"
)
print()
# Step 5: Configure Immich OAuth via API (app must already be deployed)
print("=== Configuring Immich OAuth via API ===")
print(f" Checking Immich is reachable at {immich_url} ...")
try:
immich_request(immich_url, "GET", "/api/server/ping")
except Exception as e:
print(f" ERROR: Immich app not reachable at {immich_url}")
print(f" The app must be deployed before running this script.")
print(f" Details: {e}")
sys.exit(1)
print(" Immich is reachable")
access_token = setup_immich_admin(
immich_url, args.immich_admin_email, args.immich_admin_pass,
args.immich_admin_name,
)
issuer_url = (
f"{ak_url}/application/o/{args.app_slug}"
f"/.well-known/openid-configuration"
)
configure_immich_oauth(
immich_url, access_token, issuer_url, args.client_id,
client_secret, args.app_slug,
)
print()
# Summary
print("=== Setup complete ===")
print()
print(f" Authentik domain: {args.authentik_domain}")
print(f" Immich domain: {args.immich_domain}")
print(f" Client ID: {args.client_id}")
print(f" App slug: {args.app_slug}")
print(f" Issuer URL: {issuer_url}")
if app_password:
print(f" Test user: {args.test_user}")
print(f" App password: {app_password[:10]}...")
print()
print("No redeploy needed -- Immich OAuth config takes effect immediately.")
print(f" Open https://{args.immich_domain} and click 'Login with Authentik'")
if __name__ == "__main__":
main()

429
utils/tests/helpers.py Normal file
View File

@ -0,0 +1,429 @@
"""Shared test helpers for SSO end-to-end tests.
Provides:
- abra command runner with TTY wrapper support
- HTTP helpers with retry/convergence support
- assert_converges: retry a callable until it returns truthy or timeout
- wait_for_http: poll a URL until it responds
- resolve_instance / resolve_domain / resolve_server: read from settings.toml
- load_toml_credentials: load TOML credential files
Every subprocess and HTTP call has a hard timeout to prevent hangs.
"""
import json
import os
import subprocess
import sys
import time
import tomllib
import urllib.error
import urllib.request
import urllib.parse
WORKSPACE = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
# ---------------------------------------------------------------------------
# Settings / instance resolution
# ---------------------------------------------------------------------------
def _load_settings():
"""Load and return the parsed settings.toml."""
settings_path = os.path.join(WORKSPACE, "settings.toml")
with open(settings_path, 'rb') as f:
return tomllib.load(f)
def resolve_instance():
"""Read the default instance name from settings.toml."""
settings = _load_settings()
return settings["default_instance"]
def resolve_domain(recipe):
"""Get the domain for a recipe on the active instance."""
settings = _load_settings()
instance = settings["default_instance"]
suffix = settings["instances"][instance]["domain_suffix"]
return f"{recipe}.{suffix}"
def resolve_server():
"""Get the server for the active instance."""
settings = _load_settings()
instance = settings["default_instance"]
return settings["instances"][instance]["server"]
# ---------------------------------------------------------------------------
# Credential loading
# ---------------------------------------------------------------------------
def resolve_domain_suffix():
"""Get the domain suffix for the active instance."""
settings = _load_settings()
instance = settings["default_instance"]
return settings["instances"][instance]["domain_suffix"]
def load_toml_credentials(recipe_dir, provider):
"""Load credentials from recipe-info/<recipe>/<provider>-test-credentials.<suffix>.toml.
The domain suffix is auto-resolved from settings.toml.
recipe_dir: absolute path to the recipe-info/<recipe> directory
provider: credential provider name (e.g. 'keycloak', 'authentik')
Returns: dict of credentials, or None if file doesn't exist.
"""
suffix = resolve_domain_suffix()
path = os.path.join(recipe_dir, f"{provider}-test-credentials.{suffix}.toml")
if not os.path.exists(path):
return None
with open(path, 'rb') as f:
return tomllib.load(f)
# ---------------------------------------------------------------------------
# Shell / abra command helpers
# ---------------------------------------------------------------------------
def run(cmd, check=True, timeout=120):
"""Run a shell command with a hard timeout.
Uses the Linux `timeout` command to guarantee the process tree is
killed after `timeout` seconds, even if the process ignores signals.
subprocess.run gets a slightly longer timeout as a fallback.
"""
# Wrap with Linux timeout --kill-after to hard-kill the entire process
wrapped = f"timeout --kill-after=5 {timeout} {cmd}"
print(f" $ {cmd}", flush=True)
try:
result = subprocess.run(
wrapped, shell=True, capture_output=True, text=True,
timeout=timeout + 15, # fallback: kill subprocess if Linux timeout fails
)
except subprocess.TimeoutExpired:
print(f" TIMEOUT after {timeout}s (subprocess fallback)", flush=True)
raise RuntimeError(f"Command timed out after {timeout}s: {cmd}")
if result.stdout.strip():
for line in result.stdout.strip().split("\n"):
print(f" {line}", flush=True)
if result.returncode != 0:
if result.stderr.strip():
for line in result.stderr.strip().split("\n"):
print(f" stderr: {line}", flush=True)
# exit code 124 = Linux timeout killed it
if result.returncode == 124:
print(f" TIMEOUT after {timeout}s", flush=True)
if check:
raise RuntimeError(f"Command timed out after {timeout}s: {cmd}")
elif check:
raise RuntimeError(
f"Command failed (exit {result.returncode}): {cmd}"
)
return result
def abra(args, tty_wrap=False, check=True, timeout=120):
"""Run an abra command, optionally with TTY wrapper."""
cmd = f"abra {args}"
if tty_wrap:
cmd = f'script -qefc "{cmd}" /dev/null 2>&1'
return run(cmd, check=check, timeout=timeout)
def fresh_app(recipe, server, domain, preset_secrets=None,
env_overrides=None):
"""Create a fresh app instance, cleaning up any leftovers first.
Undeploys, removes Docker secrets, volumes, and env file from previous
runs, then runs abra app new, applies env_overrides, inserts any
preset_secrets, and generates the rest.
preset_secrets: dict of {secret_name: value} to insert before
generating remaining secrets (e.g. {"admin_token": "..."}).
env_overrides: dict of {KEY: value} to set/uncomment in the .env file
after app new (e.g. {"COMPOSE_FILE": "compose.yml:compose.sso.yml"}).
"""
env_path = os.path.expanduser(
f"~/.abra/servers/{server}/{domain}.env"
)
# Undeploy if still running from a previous run
abra(f"app undeploy {domain} --no-input", check=False, timeout=60)
# Remove leftover volumes so the DB starts fresh (avoids password mismatch
# when secrets are regenerated but the old DB volume persists)
abra(f"app volume remove {domain} --force --no-input",
check=False, timeout=60)
# Remove leftover Docker secrets from the server
if os.path.exists(env_path):
abra(f"app secret remove {domain} --all --chaos --no-input",
tty_wrap=True, check=False, timeout=60)
# Remove leftover env file so app new succeeds
if os.path.exists(env_path):
print(f" Removing leftover env: {env_path}", flush=True)
os.remove(env_path)
abra(f"app new {recipe} --server {server} --domain {domain} --chaos --no-input",
timeout=60)
# Apply env overrides to the generated .env file
if env_overrides:
_apply_env_overrides(env_path, env_overrides)
# Insert preset secrets before generate so they use our known values
for name, value in (preset_secrets or {}).items():
abra(f"app secret insert {domain} {name} v1 {value} --chaos --no-input",
tty_wrap=True, check=False, timeout=60)
# Generate remaining secrets (check=False: warns if some already exist)
abra(f"app secret generate {domain} --all --chaos --no-input",
tty_wrap=True, check=False, timeout=60)
def _apply_env_overrides(env_path, overrides):
"""Set or uncomment values in an abra .env file."""
with open(env_path) as f:
lines = f.readlines()
remaining = dict(overrides)
new_lines = []
for line in lines:
stripped = line.strip()
matched = False
for key, value in list(remaining.items()):
# Match "KEY=...", "#KEY=...", or "# KEY=..."
if stripped.lstrip("#").strip().startswith(f"{key}="):
new_lines.append(f"{key}={value}\n")
remaining.pop(key)
matched = True
print(f" env: {key}={value}", flush=True)
break
if not matched:
new_lines.append(line)
# Append any keys that weren't found in the file
for key, value in remaining.items():
new_lines.append(f"{key}={value}\n")
print(f" env: {key}={value} (appended)", flush=True)
with open(env_path, "w") as f:
f.writelines(new_lines)
def deploy_and_wait(domain, server, url, label,
deploy_timeout=60, wait_max=300):
"""Fire off an abra deploy and then poll until all services are ready.
The deploy command sends the stack to Docker Swarm quickly, but
post-deploy hooks (set_admin_pass etc.) can hang for minutes.
We give the deploy command a short timeout — if it times out,
the deploy has already been submitted to Swarm.
Then we poll via SSH + `docker service ls` until all services
show full replicas (e.g. 1/1), followed by an HTTP check to
confirm the app is actually serving requests.
"""
print(f" Deploying {domain} (fire-and-poll) ...", flush=True)
abra(f"app deploy {domain} --chaos --force --no-input",
timeout=deploy_timeout, check=False)
# The Docker stack name is the domain with dots replaced by underscores
stack_prefix = domain.replace(".", "_")
# Poll replicas via SSH + docker service ls
def _all_replicas_ready():
result = run(
f"ssh {server} \"docker service ls"
f" --filter 'name={stack_prefix}'"
f" --format '{{{{.Replicas}}}}'\"",
check=False, timeout=30,
)
if result.returncode != 0:
return None
lines = [l.strip() for l in result.stdout.strip().split("\n") if l.strip()]
if not lines:
return None
for replicas in lines:
# Format is "1/1" — desired/running must match
parts = replicas.split("/")
if len(parts) != 2:
return None
if parts[0] != parts[1]:
return None
return True
assert_converges(
_all_replicas_ready,
f"{label} all replicas ready (docker service ls)",
max_wait=wait_max,
interval=15,
)
# HTTP check to confirm the app is actually serving
return wait_for_http(url, label, max_wait=120)
# ---------------------------------------------------------------------------
# Convergence helpers
# ---------------------------------------------------------------------------
def assert_converges(fn, description, max_wait=120, interval=10):
"""Retry fn() until it returns a truthy value or we time out.
fn() should return a truthy value on success, or raise / return falsy
on failure. The last return value or exception is reported on timeout.
Returns the truthy value on success.
"""
print(f" Waiting for: {description} (up to {max_wait}s) ...", flush=True)
deadline = time.time() + max_wait
last_error = None
last_result = None
attempts = 0
while time.time() < deadline:
attempts += 1
try:
result = fn()
if result:
print(
f" Converged after ~{int(time.time() - (deadline - max_wait))}s"
f" ({attempts} attempts)",
flush=True,
)
return result
last_result = result
except Exception as e:
last_error = e
time.sleep(interval)
# Timed out
detail = ""
if last_error:
detail = f" Last error: {last_error}"
elif last_result is not None:
detail = f" Last result: {last_result}"
raise RuntimeError(
f"Did not converge: {description} after {max_wait}s"
f" ({attempts} attempts).{detail}"
)
def wait_for_http(url, label, max_wait=300, interval=10):
"""Poll a URL until it returns a non-5xx response. Raises on timeout."""
def _check():
try:
req = urllib.request.Request(url, method="GET")
with urllib.request.urlopen(req, timeout=10) as resp:
code = resp.getcode()
if 200 <= code < 500:
return code
except urllib.error.HTTPError as e:
if e.code < 500:
return e.code
except Exception:
pass
return None
code = assert_converges(_check, f"{label} responding at {url}", max_wait, interval)
print(f" {label} is up (HTTP {code})", flush=True)
return code
# ---------------------------------------------------------------------------
# HTTP helpers with retry
# ---------------------------------------------------------------------------
def http_get(url, headers=None, timeout=15):
"""GET a URL, return (status_code, parsed_json_or_None).
Does NOT retry — use retry_http_get or assert_converges for that.
"""
req = urllib.request.Request(url, method="GET")
for k, v in (headers or {}).items():
req.add_header(k, v)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read()
try:
return resp.getcode(), json.loads(raw)
except (json.JSONDecodeError, ValueError):
return resp.getcode(), None
except urllib.error.HTTPError as e:
try:
raw = e.read().decode(errors="replace")
return e.code, json.loads(raw)
except Exception:
return e.code, None
except Exception:
return 0, None
def http_post(url, data=None, headers=None, content_type="application/json",
timeout=15):
"""POST to a URL, return (status_code, parsed_json_or_None).
Does NOT retry — use assert_converges for that.
"""
if content_type == "application/json" and data is not None:
body = json.dumps(data).encode()
elif content_type == "application/x-www-form-urlencoded" and data is not None:
body = urllib.parse.urlencode(data).encode()
else:
body = None
req = urllib.request.Request(url, data=body, method="POST")
req.add_header("Content-Type", content_type)
for k, v in (headers or {}).items():
req.add_header(k, v)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read()
try:
return resp.getcode(), json.loads(raw)
except (json.JSONDecodeError, ValueError):
return resp.getcode(), None
except urllib.error.HTTPError as e:
try:
raw = e.read().decode(errors="replace")
return e.code, json.loads(raw)
except Exception:
return e.code, None
except Exception:
return 0, None
def retry_http_get(url, headers=None, expect_status=200, max_wait=90,
interval=10, timeout=15):
"""GET with retries until expected status. Returns (status, json)."""
result = [None, None]
def _check():
s, j = http_get(url, headers=headers, timeout=timeout)
result[0], result[1] = s, j
return s == expect_status
assert_converges(_check, f"GET {url} -> {expect_status}", max_wait, interval)
return result[0], result[1]
def retry_http_post(url, data=None, headers=None,
content_type="application/json", expect_fn=None,
max_wait=90, interval=10, timeout=15):
"""POST with retries until expect_fn(status, json) returns truthy.
If expect_fn is None, succeeds on any 2xx.
Returns (status, json).
"""
if expect_fn is None:
expect_fn = lambda s, j: 200 <= s < 300
result = [None, None]
def _check():
s, j = http_post(url, data=data, headers=headers,
content_type=content_type, timeout=timeout)
result[0], result[1] = s, j
return expect_fn(s, j)
assert_converges(_check, f"POST {url}", max_wait, interval)
return result[0], result[1]

272
utils/tests/test_cryptpad_sso.py Executable file
View File

@ -0,0 +1,272 @@
#!/usr/bin/env python3
"""End-to-end test: create fresh Authentik + CryptPad instances, run SSO setup, verify.
Creates BRAND NEW app instances with unique domains via abra app new.
This ensures the scripts work on a completely clean install that has
never been configured before.
Usage:
python3 utils/tests/test_cryptpad_sso.py
"""
import atexit
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from helpers import (
WORKSPACE, resolve_instance, run, abra, fresh_app, deploy_and_wait,
assert_converges, http_get, http_post, retry_http_get,
)
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
INSTANCE = resolve_instance()
SERVER = f"{INSTANCE}.commoninternet.net"
AK_TEST_DOMAIN = f"ak-cpadtest.{SERVER}"
CPAD_TEST_DOMAIN = f"cpad-ssotest.{SERVER}"
AK_TEST_TOKEN = "ssotest-cpad-admin-token"
print("=== CryptPad SSO End-to-End Test ===")
print(f" Server: {SERVER}")
print(f" Authentik (new): {AK_TEST_DOMAIN}")
print(f" CryptPad (new): {CPAD_TEST_DOMAIN}")
print()
# ---------------------------------------------------------------------------
# Cleanup on exit
# ---------------------------------------------------------------------------
def cleanup():
print()
print("=== Cleanup ===")
print(" Undeploying test instances ...")
abra(f"app undeploy {CPAD_TEST_DOMAIN} --no-input", check=False, timeout=60)
abra(f"app undeploy {AK_TEST_DOMAIN} --no-input", check=False, timeout=60)
print(" Cleanup done")
atexit.register(cleanup)
failures = []
# ---------------------------------------------------------------------------
# Step 1: Create fresh Authentik
# ---------------------------------------------------------------------------
print("=== Step 1: Create fresh Authentik instance ===")
fresh_app("authentik", SERVER, AK_TEST_DOMAIN,
preset_secrets={"admin_token": AK_TEST_TOKEN})
deploy_and_wait(AK_TEST_DOMAIN, SERVER, f"https://{AK_TEST_DOMAIN}", "Authentik",
deploy_timeout=60, wait_max=300)
# The deploy post-command set_admin_pass gets killed because it runs before
# the worker has finished migrations. Re-run it with retries — the worker
# needs time to complete its bootstrap before akadmin exists.
def _set_admin_pass():
result = abra(
f"app cmd {AK_TEST_DOMAIN} worker set_admin_pass --chaos --no-input",
tty_wrap=True, check=False, timeout=120,
)
output = result.stdout if result else ""
if "Created authentik-bootstrap-token" in output or "Changed authentik-bootstrap-token" in output:
return True
if "Changed akadmin password" in output:
return True
return None
assert_converges(_set_admin_pass, "set_admin_pass (bootstrap token)",
max_wait=600, interval=15)
# Wait for the API to become ready with the admin token
ak_api = f"https://{AK_TEST_DOMAIN}/api/v3"
ak_headers = {"Authorization": f"Bearer {AK_TEST_TOKEN}"}
def _ak_api_ready():
code, body = http_get(f"{ak_api}/flows/instances/", headers=ak_headers)
if code == 200:
return True
print(f" API check: HTTP {code}", flush=True)
return False
assert_converges(_ak_api_ready, "Authentik API ready", max_wait=120)
print()
# ---------------------------------------------------------------------------
# Step 2: Create fresh CryptPad
# ---------------------------------------------------------------------------
print("=== Step 2: Create fresh CryptPad instance ===")
fresh_app("cryptpad", SERVER, CPAD_TEST_DOMAIN)
deploy_and_wait(CPAD_TEST_DOMAIN, SERVER, f"https://{CPAD_TEST_DOMAIN}", "CryptPad",
deploy_timeout=60, wait_max=180)
print()
# ---------------------------------------------------------------------------
# Step 3: Run setup script
# ---------------------------------------------------------------------------
print("=== Step 3: Run setup_cryptpad_sso.py ===")
setup_script = os.path.join(WORKSPACE, "utils", "setup_cryptpad_sso.py")
run(f"python3 {setup_script}"
f" --authentik-domain {AK_TEST_DOMAIN}"
f" --authentik-token {AK_TEST_TOKEN}"
f" --cryptpad-domain {CPAD_TEST_DOMAIN}",
timeout=120)
print()
# ---------------------------------------------------------------------------
# Step 4: Redeploy CryptPad with SSO config
# ---------------------------------------------------------------------------
print("=== Step 4: Redeploy CryptPad with SSO config ===")
deploy_and_wait(CPAD_TEST_DOMAIN, SERVER, f"https://{CPAD_TEST_DOMAIN}",
"CryptPad (post-SSO)", deploy_timeout=60, wait_max=300)
print()
# ---------------------------------------------------------------------------
# Step 5: Verify OIDC discovery endpoint (with retries)
# ---------------------------------------------------------------------------
print("=== Step 5: Verify OIDC discovery endpoint ===")
discovery_url = (
f"https://{AK_TEST_DOMAIN}/application/o/cryptpad"
f"/.well-known/openid-configuration"
)
try:
retry_http_get(discovery_url, expect_status=200, max_wait=60)
print(" PASS: OIDC discovery endpoint OK")
except RuntimeError as e:
print(f" FAIL: {e}")
failures.append("OIDC discovery")
print()
# ---------------------------------------------------------------------------
# Step 6: Obtain token from Authentik via password grant (with retries)
# ---------------------------------------------------------------------------
print("=== Step 6: Obtain token from Authentik ===")
# Get client secret — retry because Authentik may still be indexing the provider
def _get_client_secret():
_, providers = http_get(
f"{ak_api}/providers/oauth2/?search=cryptpad", headers=ak_headers,
)
if not providers:
return None
for r in providers.get("results", []):
if r["name"] == "cryptpad":
_, detail = http_get(
f"{ak_api}/providers/oauth2/{r['pk']}/", headers=ak_headers,
)
if detail:
return detail.get("client_secret")
return None
try:
client_secret = assert_converges(
_get_client_secret, "retrieve client secret", max_wait=60,
)
except RuntimeError:
client_secret = None
print(" FAIL: Could not retrieve client secret from Authentik")
failures.append("client secret retrieval")
if client_secret:
# Get app password — retry
def _get_app_password():
_, resp = http_get(
f"{ak_api}/core/tokens/testuser-app-password/view_key/",
headers=ak_headers,
)
return resp.get("key") if resp else None
try:
app_password = assert_converges(
_get_app_password, "retrieve app password", max_wait=60,
)
except RuntimeError:
app_password = None
print(" FAIL: Could not retrieve app password")
failures.append("app password retrieval")
if app_password:
# Token grant — retry (Authentik token endpoint may take a moment)
token_url = f"https://{AK_TEST_DOMAIN}/application/o/token/"
def _get_token():
_, resp = http_post(token_url, data={
"grant_type": "password",
"client_id": "cryptpad",
"client_secret": client_secret,
"username": "testuser",
"password": app_password,
"scope": "openid email profile",
}, content_type="application/x-www-form-urlencoded")
return (resp or {}).get("access_token")
try:
access_token = assert_converges(
_get_token, "password grant token", max_wait=60,
)
print(f" PASS: Got access token ({len(access_token)} chars)")
except RuntimeError:
print(" FAIL: Token request did not succeed")
failures.append("token grant")
print()
# ---------------------------------------------------------------------------
# Step 7: Verify CryptPad /ssoauth endpoint (with retries)
# ---------------------------------------------------------------------------
print("=== Step 7: Verify CryptPad /ssoauth endpoint ===")
def _check_ssoauth():
status, _ = http_get(f"https://{CPAD_TEST_DOMAIN}/ssoauth")
# Any response except 404 means the SSO plugin is loaded
if status == 404:
return None
if status == 0:
return None # connection error, retry
return status
try:
sso_status = assert_converges(
_check_ssoauth, "/ssoauth endpoint exists (not 404)", max_wait=120,
)
print(f" PASS: /ssoauth endpoint exists (HTTP {sso_status})")
except RuntimeError:
print(" FAIL: /ssoauth returned 404 -- SSO plugin may not be loaded")
failures.append("/ssoauth endpoint")
print()
# ---------------------------------------------------------------------------
# Result
# ---------------------------------------------------------------------------
if failures:
print(f"FAIL: CryptPad SSO end-to-end test failed: {', '.join(failures)}")
sys.exit(1)
else:
print("PASS: CryptPad SSO end-to-end test passed")
print(f" Fresh Authentik ({AK_TEST_DOMAIN}) + CryptPad ({CPAD_TEST_DOMAIN})")
print(" created from scratch, SSO setup, OIDC discovery OK,")
print(" token grant OK, /ssoauth endpoint exists.")

276
utils/tests/test_immich_sso.py Executable file
View File

@ -0,0 +1,276 @@
#!/usr/bin/env python3
"""End-to-end test: create fresh Authentik + Immich instances, run SSO setup, verify.
Creates BRAND NEW app instances with unique domains via abra app new.
This ensures the scripts work on a completely clean install that has
never been configured before.
Usage:
python3 utils/tests/test_immich_sso.py
"""
import atexit
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from helpers import (
WORKSPACE, resolve_instance, run, abra, fresh_app, deploy_and_wait,
assert_converges, http_get, http_post, retry_http_get,
)
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
INSTANCE = resolve_instance()
SERVER = f"{INSTANCE}.commoninternet.net"
AK_TEST_DOMAIN = f"ak-immichtest.{SERVER}"
IMMICH_TEST_DOMAIN = f"immich-ssotest.{SERVER}"
AK_TEST_TOKEN = "ssotest-immich-admin-token"
IMMICH_ADMIN_EMAIL = "admin@immich-ssotest.test"
IMMICH_ADMIN_PASS = "adminpass123"
print("=== Immich SSO End-to-End Test ===")
print(f" Server: {SERVER}")
print(f" Authentik (new): {AK_TEST_DOMAIN}")
print(f" Immich (new): {IMMICH_TEST_DOMAIN}")
print()
# ---------------------------------------------------------------------------
# Cleanup on exit
# ---------------------------------------------------------------------------
def cleanup():
print()
print("=== Cleanup ===")
print(" Undeploying test instances ...")
abra(f"app undeploy {IMMICH_TEST_DOMAIN} --no-input", check=False, timeout=60)
abra(f"app undeploy {AK_TEST_DOMAIN} --no-input", check=False, timeout=60)
print(" Cleanup done")
atexit.register(cleanup)
failures = []
# ---------------------------------------------------------------------------
# Step 1: Create fresh Authentik
# ---------------------------------------------------------------------------
print("=== Step 1: Create fresh Authentik instance ===")
fresh_app("authentik", SERVER, AK_TEST_DOMAIN,
preset_secrets={"admin_token": AK_TEST_TOKEN})
deploy_and_wait(AK_TEST_DOMAIN, SERVER, f"https://{AK_TEST_DOMAIN}", "Authentik",
deploy_timeout=60, wait_max=300)
# The deploy post-command set_admin_pass gets killed because it runs before
# the worker has finished migrations. Re-run it with retries — the worker
# needs time to complete its bootstrap before akadmin exists.
def _set_admin_pass():
result = abra(
f"app cmd {AK_TEST_DOMAIN} worker set_admin_pass --chaos --no-input",
tty_wrap=True, check=False, timeout=120,
)
output = result.stdout if result else ""
if "Created authentik-bootstrap-token" in output or "Changed authentik-bootstrap-token" in output:
return True
if "Changed akadmin password" in output:
return True
return None
assert_converges(_set_admin_pass, "set_admin_pass (bootstrap token)",
max_wait=600, interval=15)
# Wait for the API to become ready with the admin token
ak_api = f"https://{AK_TEST_DOMAIN}/api/v3"
ak_headers = {"Authorization": f"Bearer {AK_TEST_TOKEN}"}
def _ak_api_ready():
code, body = http_get(f"{ak_api}/flows/instances/", headers=ak_headers)
if code == 200:
return True
print(f" API check: HTTP {code}", flush=True)
return False
assert_converges(_ak_api_ready, "Authentik API ready", max_wait=120)
print()
# ---------------------------------------------------------------------------
# Step 2: Create fresh Immich
# ---------------------------------------------------------------------------
print("=== Step 2: Create fresh Immich instance ===")
fresh_app("immich", SERVER, IMMICH_TEST_DOMAIN)
deploy_and_wait(IMMICH_TEST_DOMAIN, SERVER, f"https://{IMMICH_TEST_DOMAIN}", "Immich",
deploy_timeout=60, wait_max=180)
# Wait for Immich API to be ready (not just the web UI)
assert_converges(
lambda: http_get(f"https://{IMMICH_TEST_DOMAIN}/api/server/ping")[0] == 200,
"Immich API ready",
max_wait=120,
)
print()
# ---------------------------------------------------------------------------
# Step 3: Run setup script
# ---------------------------------------------------------------------------
print("=== Step 3: Run setup_immich_sso.py ===")
setup_script = os.path.join(WORKSPACE, "utils", "setup_immich_sso.py")
run(f"python3 {setup_script}"
f" --authentik-domain {AK_TEST_DOMAIN}"
f" --authentik-token {AK_TEST_TOKEN}"
f" --immich-domain {IMMICH_TEST_DOMAIN}"
f" --immich-admin-email {IMMICH_ADMIN_EMAIL}"
f" --immich-admin-pass {IMMICH_ADMIN_PASS}",
timeout=120)
print()
# ---------------------------------------------------------------------------
# Step 4: Verify OIDC discovery endpoint (with retries)
# ---------------------------------------------------------------------------
print("=== Step 4: Verify OIDC discovery endpoint ===")
discovery_url = (
f"https://{AK_TEST_DOMAIN}/application/o/immich"
f"/.well-known/openid-configuration"
)
try:
retry_http_get(discovery_url, expect_status=200, max_wait=60)
print(" PASS: OIDC discovery endpoint OK")
except RuntimeError as e:
print(f" FAIL: {e}")
failures.append("OIDC discovery")
print()
# ---------------------------------------------------------------------------
# Step 5: Obtain token from Authentik via password grant (with retries)
# ---------------------------------------------------------------------------
print("=== Step 5: Obtain token from Authentik ===")
# Get client secret — retry
def _get_client_secret():
_, providers = http_get(
f"{ak_api}/providers/oauth2/?search=immich", headers=ak_headers,
)
if not providers:
return None
for r in providers.get("results", []):
if r["name"] == "immich":
_, detail = http_get(
f"{ak_api}/providers/oauth2/{r['pk']}/", headers=ak_headers,
)
if detail:
return detail.get("client_secret")
return None
try:
client_secret = assert_converges(
_get_client_secret, "retrieve client secret", max_wait=60,
)
except RuntimeError:
client_secret = None
print(" FAIL: Could not retrieve client secret from Authentik")
failures.append("client secret retrieval")
if client_secret:
# Get app password — retry
def _get_app_password():
_, resp = http_get(
f"{ak_api}/core/tokens/testuser-app-password/view_key/",
headers=ak_headers,
)
return resp.get("key") if resp else None
try:
app_password = assert_converges(
_get_app_password, "retrieve app password", max_wait=60,
)
except RuntimeError:
app_password = None
print(" FAIL: Could not retrieve app password")
failures.append("app password retrieval")
if app_password:
# Token grant — retry
token_url = f"https://{AK_TEST_DOMAIN}/application/o/token/"
def _get_token():
_, resp = http_post(token_url, data={
"grant_type": "password",
"client_id": "immich",
"client_secret": client_secret,
"username": "testuser",
"password": app_password,
"scope": "openid email profile",
}, content_type="application/x-www-form-urlencoded")
return (resp or {}).get("access_token")
try:
access_token = assert_converges(
_get_token, "password grant token", max_wait=60,
)
print(f" PASS: Got access token ({len(access_token)} chars)")
except RuntimeError:
print(" FAIL: Token request did not succeed")
failures.append("token grant")
print()
# ---------------------------------------------------------------------------
# Step 6: Verify Immich OAuth authorize endpoint (with retries)
# ---------------------------------------------------------------------------
print("=== Step 6: Verify Immich /api/oauth/authorize ===")
def _check_authorize():
_, resp = http_post(
f"https://{IMMICH_TEST_DOMAIN}/api/oauth/authorize",
data={"redirectUri": f"https://{IMMICH_TEST_DOMAIN}/auth/login"},
)
url = (resp or {}).get("url", "")
return url if url else None
try:
redirect_url = assert_converges(
_check_authorize, "OAuth authorize returns redirect URL", max_wait=60,
)
if AK_TEST_DOMAIN in redirect_url:
print(" PASS: OAuth authorize returns Authentik redirect URL")
else:
print(f" WARN: Redirect URL doesn't contain Authentik domain: {redirect_url}")
print(" (May still be OK if using a different OIDC discovery URL)")
except RuntimeError:
print(" FAIL: /api/oauth/authorize did not return a redirect URL")
failures.append("OAuth authorize")
print()
# ---------------------------------------------------------------------------
# Result
# ---------------------------------------------------------------------------
if failures:
print(f"FAIL: Immich SSO end-to-end test failed: {', '.join(failures)}")
sys.exit(1)
else:
print("PASS: Immich SSO end-to-end test passed")
print(f" Fresh Authentik ({AK_TEST_DOMAIN}) + Immich ({IMMICH_TEST_DOMAIN})")
print(" created from scratch, SSO setup, OIDC discovery OK,")
print(" token grant OK, OAuth authorize endpoint returns redirect URL.")