KNEL/agent-identity-provisioning
5
0
Fork 0
Code Pull Requests Actions Packages Releases Activity
Files
5e1d043890041743cffad8478b8f1856acf71a57
agent-identity-provisioning/provision-agent.py
T
TSYS Group COO 6d89f16610 feat: consolidate proven Discourse/Redmine flows into provision-agent.py 
Replaced the stale session-1 provision_discourse/provision_redmine with
the flows proven on vp-techops this session. Both are now parameterized
by the agent manifest (username derived from name, hyphens stripped,
overridable via username: field).

Discourse: login modal -> OpenID button -> signup on first login ->
RSA User API key flow (PKCS1v15 decrypt, JSON payload).

Redmine: KNEL Cloud SSO button -> consent -> Show/Reset on the API
access key section via targeted DOM traversal.

Added docs/JOURNAL.md with all working selectors, flows, gotchas, and
verification results so future sessions do not rediscover them.
2026-08-14 08:23:37 -05:00

1180 lines
44 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
provision-agent.py — Playwright automation for AI agent identity provisioning.
Enrolls AI agent identities in Cloudron, logs into Gitea/Discourse/Redmine via
SSO, generates API keys, and stores all credentials in Bitwarden.
Usage:
python3 provision-agent.py # provision all agents in manifest
python3 provision-agent.py --agent vp-techops # provision one agent
python3 provision-agent.py --phase1-only # Cloudron enrollment only
python3 provision-agent.py --dry-run # validate manifest without browser
Manifest: agents.yaml (see agents.yaml.example)
See: ~/Q3/agent-identity-bootstrap.md for the full architecture.
"""
import argparse
import json
import logging
import os
import re
import sys
import time
from pathlib import Path
import yaml
from playwright.sync_api import BrowserContext, Page, sync_playwright
from bw_helper import BitwardenHelper
# ---------------------------------------------------------------------------
# Logging
# ---------------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%H:%M:%S",
)
log = logging.getLogger("provision")
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
CLOUDRON_BASE = os.environ.get("CLOUDRON_BASE", "https://my.knownelement.com")
GITEA_URL = os.environ.get("GITEA_URL", "https://git.knownelement.com")
DISCOURSE_URL = os.environ.get("DISCOURSE_URL", "https://community.turnsys.com")
REDMINE_URL = os.environ.get("REDMINE_URL", "https://projects.knownelement.com")
STATE_DIR = Path("/app/state")
def _debug_dump(page: Page, label: str) -> None:
"""Save screenshot + DOM dump for debugging selector issues."""
STATE_DIR.mkdir(parents=True, exist_ok=True)
ts = time.strftime("%H%M%S")
screenshot_path = STATE_DIR / f"debug-{label}-{ts}.png"
dom_path = STATE_DIR / f"debug-{label}-{ts}.txt"
try:
page.screenshot(path=str(screenshot_path), full_page=True)
except Exception:
pass
try:
elements = page.evaluate("""() => {
const results = [];
document.querySelectorAll('input, button, select, textarea, label, h2, h3, h4, h5, code, .ui.header, .ui.message, [data-tab], .tw-font-semibold').forEach(el => {
const tag = el.tagName.toLowerCase();
const text = el.textContent.trim().substring(0, 80);
const id = el.id || '';
const name = el.getAttribute('name') || '';
const type = el.getAttribute('type') || '';
const value = el.getAttribute('value') || '';
const cls = (el.className || '').substring(0, 50);
if (text || id || name || type || value) results.push(`<${tag}> id=${id} name=${name} type=${type} value=${value} class=${cls} text="${text}"`);
});
return results.join('\\n');
}""")
dom_path.write_text(f"URL: {page.url}\\n\\n{elements}")
log.info(f" Debug dump saved: {screenshot_path.name}, {dom_path.name}")
except Exception as e:
log.warning(f" Debug dump failed: {e}")
# ---------------------------------------------------------------------------
# Cloudron enrollment
# ---------------------------------------------------------------------------
def enroll_cloudron(
page: Page,
agent: dict,
bw: BitwardenHelper,
) -> dict:
"""
Phase 1: Accept Cloudron invite, set password, enable 2FA.
Returns a dict with the agent's Cloudron credentials.
"""
name = agent["name"]
invite_url = agent["cloudron_invite"]
display_name = agent.get("display_name", name)
log.info(f"[{name}] Phase 1: Cloudron enrollment — {invite_url}")
# Check if already provisioned
item_name = f"{name} Cloudron"
if bw.item_exists(item_name):
log.info(f"[{name}] Cloudron credential already exists in Bitwarden — skipping")
cloudron_email = agent.get("cloudron_email", f"{name}@turnsys.com")
return {
"username": cloudron_email,
"password": bw.get_item_password(item_name),
}
# Generate a strong password
password = bw.generate_password(length=32)
log.info(f"[{name}] Generated password ({len(password)} chars)")
# Navigate to invite link
page.goto(invite_url, wait_until="networkidle")
page.wait_for_timeout(2000)
# Fill in the invite acceptance form using real Cloudron selectors
page.wait_for_selector('#inputPassword', timeout=15000)
page.fill('#inputPassword', password)
page.fill('#inputPasswordRepeat', password)
# Set display name if field exists
display_field = page.query_selector('#inputDisplayName')
if display_field:
display_field.fill(display_name)
# Submit — Cloudron uses a button with class btn-primary or type submit
submit = page.query_selector('button[type="submit"], button.btn-primary, button:has-text("Setup"), button:has-text("Create"), button:has-text("Accept")')
if submit:
submit.click()
else:
page.keyboard.press('Enter')
page.wait_for_load_state("networkidle")
log.info(f"[{name}] Invite accepted")
# Enable 2FA
totp_secret = enable_cloudron_2fa(page, agent, bw)
# Store credential in Bitwarden
cloudron_email = agent.get("cloudron_email", f"{name}@turnsys.com")
bw.create_item(
name=item_name,
username=cloudron_email,
password=password,
uris=[CLOUDRON_BASE],
collection_name=name,
totp_secret=totp_secret,
)
log.info(f"[{name}] Cloudron credential stored in Bitwarden (collection: {name})")
return {"username": cloudron_email, "password": password, "totp_secret": totp_secret}
def enable_cloudron_2fa(page: Page, agent: dict, bw: BitwardenHelper) -> str:
"""
Enable TOTP 2FA on a Cloudron account via the profile page.
Cloudron's 2FA enrollment flow (discovered via DOM dump):
1. Navigate to #/profile
2. Click "Setup" to start 2FA enrollment
3. Cloudron defaults to Passkey -- click "switchToTotp" to switch
4. TOTP secret appears in page text (base32 encoded)
5. Enter TOTP code in #totpTokenInput, click Enable
Returns the TOTP secret (empty string if already enabled or failed).
"""
name = agent["name"]
log.info(f"[{name}] Enabling 2FA on Cloudron account")
page.evaluate('() => window.location.hash = "#/profile"')
page.wait_for_timeout(3000)
# If profile didn't load, try clicking the profile nav link
if "#/profile" not in page.url:
for nav_text in ["Profile", "Settings", "Account"]:
sec_link = page.locator(f'a[href*="#/profile"], a:has-text("{nav_text}")')
if sec_link.count() > 0:
sec_link.first.click()
page.wait_for_timeout(3000)
break
_debug_dump(page, f"cloudron-profile-{name}")
# Check if 2FA is already enabled
page_text = page.evaluate("() => document.body.innerText")
if "totp" in page_text.lower() and "enabled" in page_text.lower():
log.info(f"[{name}] 2FA already enabled on Cloudron")
return ""
# Step 1: Click "Setup" to start 2FA enrollment
setup_clicked = False
for selector in [
'text=Setup',
'[role="button"]:has-text("Setup")',
'button:has-text("Setup")',
'a:has-text("Setup")',
]:
loc = page.locator(selector)
if loc.count() > 0 and loc.first.is_visible():
loc.first.click()
page.wait_for_timeout(3000)
setup_clicked = True
break
if not setup_clicked:
log.warning(f"[{name}] Could not find 2FA Setup button")
_debug_dump(page, f"cloudron-2fa-no-setup-{name}")
return ""
# Step 2: Switch from Passkey to TOTP mode
page_text = page.evaluate("() => document.body.innerText")
if "switchToTotp" in page_text or "Passkey" in page_text:
for selector in [
'text=switchToTotp',
'text=profile.enable2FA.switchToTotp',
'a:has-text("TOTP")',
'[role="button"]:has-text("TOTP")',
]:
loc = page.locator(selector)
if loc.count() > 0 and loc.first.is_visible():
loc.first.click()
page.wait_for_timeout(3000)
break
_debug_dump(page, f"cloudron-2fa-totp-mode-{name}")
# Step 3: Extract TOTP secret (base32 string in page text)
import re
page_text = page.evaluate("() => document.body.innerText")
totp_secret = ""
# Look in code/pre elements first
for el in page.query_selector_all("code, pre"):
text = el.text_content().strip().replace(" ", "")
if text and re.match(r'^[A-Z2-7=]+$', text):
totp_secret = text
break
# Fall back to regex in page text
if not totp_secret:
matches = re.findall(r'[A-Z2-7]{16,}=?', page_text.replace(" ", ""))
if matches:
totp_secret = matches[0]
if not totp_secret:
log.error(f"[{name}] Could not extract TOTP secret")
_debug_dump(page, f"cloudron-2fa-no-secret-{name}")
return ""
log.info(f"[{name}] Extracted TOTP secret: {totp_secret[:4]}...")
# Step 4: Enter TOTP confirmation code
import pyotp
totp_code = pyotp.TOTP(totp_secret).now()
token_input = page.query_selector("#totpTokenInput")
if not token_input:
for sel in ['input[name="totpToken"]', 'input[type="text"]:visible']:
loc = page.locator(sel)
if loc.count() > 0 and loc.first.is_visible():
token_input = loc.first.element_handle()
break
if token_input:
token_input.click()
page.keyboard.type(totp_code)
page.wait_for_timeout(500)
# Click Enable/Confirm button
for btn_text in ["Enable", "Confirm", "Verify", "OK", "Save"]:
loc = page.locator(f'[role="button"]:has-text("{btn_text}"), button:has-text("{btn_text}")')
if loc.count() > 0 and loc.first.is_visible():
loc.first.click()
break
page.wait_for_timeout(3000)
log.info(f"[{name}] 2FA confirmation submitted")
else:
log.warning(f"[{name}] Could not find TOTP confirmation input")
_debug_dump(page, f"cloudron-2fa-no-input-{name}")
return totp_secret
def decode_qr_from_base64(data_uri: str) -> str:
"""Decode a TOTP secret from a base64 QR code data URI."""
import base64
import io
from PIL import Image
from pyzbar.pyzbar import decode
# Extract base64 data from data URI
header, b64data = data_uri.split(",", 1)
img_bytes = base64.b64decode(b64data)
img = Image.open(io.BytesIO(img_bytes))
decoded = decode(img)
if decoded:
# TOTP URIs look like: otpauth://totp/Label?secret=XXXX&...
uri = decoded[0].data.decode()
if "secret=" in uri:
return uri.split("secret=")[1].split("&")[0]
raise RuntimeError("Could not decode TOTP secret from QR code")
# ---------------------------------------------------------------------------
# System access (Phase 2)
# ---------------------------------------------------------------------------
def cloudron_panel_login(page: Page, agent: dict, bw: BitwardenHelper) -> bool:
"""
Login to the Cloudron panel to establish an SSO session.
Once authenticated to the Cloudron panel, subsequent SSO flows to
Cloudron-managed apps (Gitea, Redmine, Discourse, etc.) auto-approve
without requiring a separate consent click.
Returns True if login succeeded or session already active.
"""
name = agent["name"]
cloudron_email = agent.get("cloudron_email", f"{name}@turnsys.com")
cloudron_item = f"{name} Cloudron"
password = bw.get_item_password(cloudron_item)
log.info(f"[{name}] Establishing Cloudron panel session")
page.goto(f"{CLOUDRON_BASE}/login.html", wait_until="networkidle", timeout=15000)
page.wait_for_timeout(2000)
# Already logged in? (must be on the panel, not redirected to OIDC)
if "login" not in page.url.lower() and "openid" not in page.url.lower():
log.info(f"[{name}] Cloudron panel session already active")
return True
# If redirected to OIDC login page, fill that too
# (Cloudron login.html and OIDC login look similar, both use #inputUsername/#inputPassword)
# Fill Cloudron login form (Pankow/Vue needs keyboard events, not fill())
page.wait_for_selector("#inputPassword", timeout=15000)
page.click("#inputUsername")
page.keyboard.type(cloudron_email)
page.click("#inputPassword")
page.keyboard.type(password)
# Submit: Cloudron Pankow UI uses <div role="button"> instead of <button>.
submitted = False
for text in ["Log in", "Sign in", "Submit", "Continue"]:
btn = page.locator(f'[role="button"]:has-text("{text}"), button:has-text("{text}")')
if btn.count() > 0 and btn.first.is_visible():
btn.first.click()
submitted = True
break
if not submitted:
page.evaluate("() => { const f = document.querySelector('form'); if (f) f.requestSubmit(); }")
submitted = True
page.wait_for_timeout(5000)
# Handle TOTP if prompted (Cloudron OIDC uses #inputTotpToken, not #inputTotp)
totp_input = page.query_selector("#inputTotpToken")
if totp_input and totp_input.is_visible():
totp_code = bw.get_totp(cloudron_item)
totp_input.click()
page.keyboard.type(totp_code)
# Cloudron TOTP submit button has specific ID
submit = page.locator('#totpTokenSubmitButton')
if submit.count() == 0:
for text in ["Log in", "Sign in", "Submit", "Continue"]:
submit = page.locator(f'[role="button"]:has-text("{text}"), button:has-text("{text}")')
if submit.count() > 0 and submit.first.is_visible():
break
if submit.count() > 0:
submit.first.click()
else:
page.evaluate("() => { const f = document.querySelector('form'); if (f) f.requestSubmit(); }")
page.wait_for_timeout(5000)
# After submit, check if we're past the login form.
# The OIDC flow stays on /openid/interaction/... but changes from
# login form to consent page. Check for absence of password field.
page.wait_for_timeout(3000)
has_password = page.query_selector('#inputPassword')
if not has_password or not has_password.is_visible():
log.info(f"[{name}] Cloudron panel login successful")
return True
# Handle consent/authorize page if present
for consent_text in ["Continue", "Authorize", "Allow", "Accept"]:
consent_btn = page.locator(f'[role="button"]:has-text("{consent_text}"), button:has-text("{consent_text}")')
if consent_btn.count() > 0 and consent_btn.first.is_visible():
consent_btn.first.click()
page.wait_for_timeout(5000)
break
has_password = page.query_selector('#inputPassword')
if not has_password or not has_password.is_visible():
log.info(f"[{name}] Cloudron panel login successful")
return True
log.error(f"[{name}] Cloudron panel login failed (still on login page: {page.url})")
return False
def sso_login(page: Page, system_url: str, agent: dict, bw: BitwardenHelper,
sso_button_selector: str = "") -> bool:
"""
Log into a Cloudron-managed app via SSO.
Requires an active Cloudron panel session (call cloudron_panel_login first).
Clicks the SSO/OAuth button on the app's login page, which redirects
through Cloudron's OIDC provider and back to the app authenticated.
Args:
sso_button_selector: CSS selector for the SSO button. If empty,
tries common patterns (oauth2 link, "cloudron" text, etc.).
Returns True if login succeeded.
"""
name = agent["name"]
log.info(f"[{name}] SSO login: {system_url}")
page.goto(system_url, wait_until="networkidle")
page.wait_for_timeout(2000)
# Already authenticated?
if not page.query_selector('input[type="password"]'):
log.info(f"[{name}] Already authenticated at {system_url}")
return True
# Click the SSO/OAuth button
sso_clicked = False
if sso_button_selector:
btn = page.locator(sso_button_selector)
if btn.count() > 0:
btn.first.click()
sso_clicked = True
if not sso_clicked:
for selector in [
'a[href*="oauth2/cloudron"]',
'a[href*="oauth"]',
'button[id*="login-oauth"]',
'a:has-text("cloudron")',
'a:has-text("Cloud")',
'button:has-text("cloudron")',
'button:has-text("Cloud")',
]:
btn = page.locator(selector)
if btn.count() > 0 and btn.first.is_visible():
btn.first.click()
sso_clicked = True
break
if not sso_clicked:
log.error(f"[{name}] Could not find SSO button at {system_url}")
_debug_dump(page, f"sso-no-button-{name}")
return False
page.wait_for_timeout(5000)
log.info(f"[{name}] SSO redirect: {page.url}")
# Handle Cloudron OIDC login page if redirected.
# The OIDC page may show: (a) a login form, (b) a consent page,
# or (c) nothing visible (auto-redirect). Handle all three.
if "openid" in page.url or "interaction" in page.url:
page.wait_for_timeout(2000)
has_login_form = page.query_selector("#inputPassword")
if has_login_form and has_login_form.is_visible():
log.info(f"[{name}] Handling Cloudron OIDC login")
cloudron_email = agent.get("cloudron_email", f"{name}@turnsys.com")
cloudron_item = f"{name} Cloudron"
password = bw.get_item_password(cloudron_item)
page.click("#inputUsername")
page.keyboard.type(cloudron_email)
page.click("#inputPassword")
page.keyboard.type(password)
for text in ["Log in", "Sign in", "Submit", "Continue"]:
btn = page.locator(f'[role="button"]:has-text("{text}"), button:has-text("{text}")')
if btn.count() > 0 and btn.first.is_visible():
btn.first.click()
break
else:
page.evaluate("() => { const f = document.querySelector('form'); if (f) f.requestSubmit(); }")
page.wait_for_timeout(5000)
# Handle TOTP if prompted
totp_input = page.query_selector("#inputTotp")
if totp_input and totp_input.is_visible():
totp_code = bw.get_totp(cloudron_item)
totp_input.fill(totp_code)
for text in ["Log in", "Sign in", "Submit", "Continue"]:
btn = page.locator(f'[role="button"]:has-text("{text}"), button:has-text("{text}")')
if btn.count() > 0 and btn.first.is_visible():
btn.first.click()
break
page.wait_for_timeout(5000)
else:
log.info(f"[{name}] OIDC session active — looking for consent")
# Try to find and click any consent/authorize button.
# The OIDC page may require explicit consent even when authenticated.
page.wait_for_timeout(2000)
for consent_text in ["Continue", "Authorize", "Allow", "Accept", "Approve"]:
consent_btn = page.locator(f'[role="button"]:has-text("{consent_text}"), button:has-text("{consent_text}"), input[value="{consent_text}"]')
if consent_btn.count() > 0:
log.info(f"[{name}] Clicking OIDC consent: {consent_text}")
consent_btn.first.click(force=True)
page.wait_for_timeout(5000)
break
else:
# No consent button found — try submitting any form on the page
# (OIDC auto-approve may need a form POST)
page.evaluate("() => { const f = document.querySelector('form'); if (f) f.requestSubmit(); }")
page.wait_for_timeout(5000)
page.wait_for_timeout(3000)
log.info(f"[{name}] SSO result: {page.url}")
# Verify we're back on the target app (not stuck on OIDC/login)
if "openid" in page.url or "interaction" in page.url:
log.error(f"[{name}] SSO login failed (stuck on OIDC page)")
_debug_dump(page, f"sso-stuck-{name}")
return False
if "login" in page.url.lower():
log.error(f"[{name}] SSO login failed (still on login page)")
_debug_dump(page, f"sso-failed-{name}")
return False
log.info(f"[{name}] SSO login successful")
return True
def provision_gitea(page: Page, agent: dict, bw: BitwardenHelper) -> str:
"""Generate a Gitea API token via SSO login. Returns the token."""
name = agent["name"]
systems = agent.get("systems", {})
gitea_cfg = systems.get("gitea", {})
if not gitea_cfg:
log.info(f"[{name}] No Gitea config — skipping")
return ""
item_name = f"{name} Gitea"
if bw.item_exists(item_name):
log.info(f"[{name}] Gitea token already exists — skipping")
return bw.get_item_password(item_name)
url = gitea_cfg.get("url", GITEA_URL)
token_name = gitea_cfg.get("token_name", f"{name}-api")
sso_login(page, f"{url}/user/login", agent, bw,
sso_button_selector='a[href*="oauth2/cloudron"]')
# Navigate to API token settings
page.goto(f"{url}/user/settings/applications", wait_until="domcontentloaded")
page.wait_for_timeout(3000)
_debug_dump(page, f"gitea-settings-{name}")
# Generate new token — the input may not be visible to Playwright
# (behind flash message or in collapsed section). Use JS to fill.
page.evaluate(f"""() => {{
const input = document.querySelector('#name');
if (input) {{
input.value = '{token_name}';
input.dispatchEvent(new Event('input', {{bubbles: true}}));
input.dispatchEvent(new Event('change', {{bubbles: true}}));
}}
}}""")
# Select scopes via JS (radio buttons may also be non-visible)
scope_map = {
"repository": "write:repository",
"user": "write:user",
"organization": "write:organization",
"issue": "write:issue",
"package": "write:package",
"notification": "read:notification",
"misc": "read:misc",
}
page.evaluate("""(scopes) => {
for (const [cat, val] of Object.entries(scopes)) {
const radio = document.querySelector('input[value="' + val + '"]');
if (radio) { radio.checked = true; radio.dispatchEvent(new Event('change', {bubbles: true})); }
}
}""", scope_map)
# Click Generate Token via JS
page.evaluate("""() => {
const btns = document.querySelectorAll('button');
for (const b of btns) {
if (b.textContent.trim() === 'Generate Token') { b.click(); return; }
}
}""")
page.wait_for_timeout(3000)
# Extract the generated token.
# Gitea shows it in a flash-info message div as plain text (the token
# string itself, not wrapped in <code>).
token_el = page.query_selector('.ui.info.message.flash-info, .ui.info.message')
if not token_el:
token_el = page.query_selector('.ui.message code, input[readonly]')
token = ""
if token_el:
token = token_el.text_content().strip()
# The flash message may contain extra text; extract just the token
# (Gitea tokens are 40-char hex strings)
import re
match = re.search(r'[a-f0-9]{40}', token)
if match:
token = match.group(0)
if not token:
log.error(f"[{name}] Could not extract Gitea API token")
_debug_dump(page, f"gitea-no-token-{name}")
return ""
log.info(f"[{name}] Gitea token generated: {token[:8]}...")
# Store in Bitwarden
bw.create_item(
name=item_name,
username=name,
password=token,
uris=[url],
collection_name=name,
custom_fields={"token_name": token_name},
)
log.info(f"[{name}] Gitea token stored in Bitwarden")
return token
def provision_discourse(page: Page, agent: dict, bw: BitwardenHelper) -> str:
"""
Discourse SSO login + User API key generation.
Proven flow (session 2, verified on vp-techops):
1. Click .login-button to open the login modal
2. Click the OpenID Connect button inside the modal
3. First login redirects to /signup with email pre-authenticated:
fill username, click Sign Up
4. User API key via the RSA-encrypted flow:
- Generate RSA keypair, pass public key to /user-api-key/new
- Click Authorize
- Capture the POST response payload, decrypt (PKCS1v15),
parse JSON to extract the "key" field
The resulting key authenticates via the User-Api-Key header
(NOT Api-Key -- that is for admin-created keys).
Returns the API key (empty string on failure).
"""
name = agent["name"]
systems = agent.get("systems", {})
discourse_cfg = systems.get("discourse", {})
if not discourse_cfg:
log.info(f"[{name}] No Discourse config -- skipping")
return ""
item_name = f"{name} Discourse"
if bw.item_exists(item_name):
log.info(f"[{name}] Discourse key already exists -- skipping")
return bw.get_item_password(item_name)
url = discourse_cfg.get("url", DISCOURSE_URL)
username = agent.get("username", name.replace("-", ""))
# --- SSO login ---
page.goto(f"{url}/", wait_until="domcontentloaded", timeout=30000)
page.wait_for_timeout(3000)
if not page.query_selector("#current-user, .current-user"):
login_btn = page.locator(".login-button, button:has-text('Log In')")
if login_btn.count() > 0:
login_btn.first.click()
page.wait_for_timeout(3000)
sso_btn = page.locator('button:has-text("OpenID")')
if sso_btn.count() > 0:
sso_btn.first.click()
page.wait_for_timeout(5000)
# First login: /signup with email already authenticated by OIDC
if "/signup" in page.url:
page.wait_for_timeout(2000)
username_input = page.locator('#new-account-username, input[name="username"]')
if username_input.count() > 0 and username_input.first.is_visible():
username_input.first.click()
page.keyboard.type(username)
page.wait_for_timeout(1000)
for btn_text in ["Create Account", "Sign Up", "Register"]:
loc = page.locator(f'button:has-text("{btn_text}")')
if loc.count() > 0 and loc.first.is_visible():
loc.first.click()
page.wait_for_timeout(5000)
break
log.info(f"[{name}] Discourse account created: {username}")
page.goto(f"{url}/", wait_until="domcontentloaded", timeout=30000)
page.wait_for_timeout(3000)
if not page.query_selector("#current-user, .current-user"):
log.error(f"[{name}] Discourse SSO login failed")
_debug_dump(page, f"discourse-login-failed-{name}")
return ""
log.info(f"[{name}] Discourse SSO login OK")
# --- User API key (RSA flow) ---
import base64
import secrets
import uuid as uuid_lib
from urllib.parse import quote_plus
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding as asym_padding
from cryptography.hazmat.primitives.asymmetric import rsa as asym_rsa
private_key = asym_rsa.generate_private_key(public_exponent=65537, key_size=2048)
public_pem = private_key.public_key().public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo,
).decode("ascii")
nonce = secrets.token_hex(16)
client_id = str(uuid_lib.uuid4())
app_name = f"TSG-Agent-{agent.get('display_name', name)}"
params = (
f"?application_name={quote_plus(app_name)}"
f"&client_id={client_id}"
f"&nonce={nonce}"
f"&scopes=read%2Cwrite"
f"&public_key={quote_plus(public_pem)}"
)
api_responses = []
def handle_response(response):
if "user-api-key" in response.url and response.request.method == "POST":
try:
api_responses.append(response.text())
except Exception:
pass
page.on("response", handle_response)
try:
page.goto(f"{url}/user-api-key/new{params}", wait_until="domcontentloaded", timeout=30000)
page.wait_for_timeout(3000)
for btn_text in ["Authorize", "Approve", "Continue"]:
loc = page.locator(f'button:has-text("{btn_text}"), .btn-primary')
if loc.count() > 0 and loc.first.is_visible():
loc.first.click()
break
page.wait_for_timeout(3000)
finally:
page.remove_listener("response", handle_response)
api_key = ""
for resp_text in api_responses:
try:
data = json.loads(resp_text)
encrypted_raw = data.get("key") or data.get("payload") or ""
if not encrypted_raw:
continue
encrypted = base64.b64decode(
encrypted_raw.replace("\n", "").replace("\r", "").replace(" ", "")
)
# Discourse uses PKCS1v15; try OAEP variants as fallback
paddings = [
asym_padding.PKCS1v15(),
asym_padding.OAEP(
mgf=asym_padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(), label=None),
asym_padding.OAEP(
mgf=asym_padding.MGF1(algorithm=hashes.SHA1()),
algorithm=hashes.SHA1(), label=None),
]
for pad in paddings:
try:
decrypted = private_key.decrypt(encrypted, pad).decode("ascii")
try:
api_key = json.loads(decrypted).get("key", decrypted)
except json.JSONDecodeError:
api_key = decrypted
break
except Exception:
continue
if api_key:
break
except Exception:
continue
if not api_key:
log.error(f"[{name}] Could not extract Discourse API key")
_debug_dump(page, f"discourse-no-key-{name}")
return ""
log.info(f"[{name}] Discourse API key obtained: {api_key[:8]}...")
bw.create_item(
name=item_name,
username=username,
password=api_key,
uris=[url],
collection_name=name,
)
log.info(f"[{name}] Discourse API key stored in Bitwarden")
return api_key
def provision_redmine(page: Page, agent: dict, bw: BitwardenHelper) -> str:
"""
Redmine SSO login + API access key extraction.
Proven flow (session 2, verified on vp-techops):
1. Click "Continue with KNEL Cloud" (#login-oauth-submit-1)
2. OIDC may show a consent page -- click Continue
3. On /my/account click "Show" in the .api-key-actions section
4. Read the 40-hex key from the #api-access-key pre element
5. If no key exists yet, click the API-key Reset link via targeted
DOM traversal (a generic "Reset" match clicks the wrong section)
Prereq: the Cloudron user must have Redmine app access granted by
the Cloudron admin, else OIDC shows "You do not have access".
Returns the API key (empty string on failure).
"""
name = agent["name"]
systems = agent.get("systems", {})
redmine_cfg = systems.get("redmine", {})
if not redmine_cfg:
log.info(f"[{name}] No Redmine config -- skipping")
return ""
item_name = f"{name} Redmine"
if bw.item_exists(item_name):
log.info(f"[{name}] Redmine key already exists -- skipping")
return bw.get_item_password(item_name)
url = redmine_cfg.get("url", REDMINE_URL)
username = agent.get("username", name.replace("-", ""))
# --- SSO login ---
page.goto(f"{url}/login", wait_until="domcontentloaded", timeout=30000)
page.wait_for_timeout(3000)
sso_btn = page.locator(
'#login-oauth-submit-1, button:has-text("KNEL"), button:has-text("Continue")'
)
if sso_btn.count() > 0 and sso_btn.first.is_visible():
sso_btn.first.click()
page.wait_for_timeout(5000)
if "openid" in page.url.lower():
for consent in ["Continue", "Authorize", "Allow", "Accept"]:
cbtn = page.locator(
f'[role="button"]:has-text("{consent}"), button:has-text("{consent}")'
)
if cbtn.count() > 0 and cbtn.first.is_visible():
cbtn.first.click()
page.wait_for_timeout(5000)
break
if "/login" in page.url:
log.error(f"[{name}] Redmine SSO failed (still on login page)")
_debug_dump(page, f"redmine-sso-failed-{name}")
return ""
log.info(f"[{name}] Redmine SSO login OK")
# --- API key ---
page.goto(f"{url}/my/account", wait_until="domcontentloaded", timeout=30000)
page.wait_for_timeout(3000)
_debug_dump(page, f"redmine-account-{name}")
api_key = ""
# A key usually exists (auto-created); reveal it via "Show"
show_btn = page.locator('.api-key-actions a:has-text("Show"), a:has-text("Show")')
if show_btn.count() > 0 and show_btn.first.is_visible():
show_btn.first.click()
page.wait_for_timeout(2000)
api_el = page.query_selector("#api-access-key")
if api_el:
matches = re.findall(r"[a-f0-9]{40}", api_el.text_content())
if matches:
api_key = matches[0]
# No key yet: generate via the Reset link next to #api-access-key
if not api_key:
reset_clicked = page.evaluate("""() => {
const apiSection = document.querySelector('#api-access-key');
if (!apiSection) return false;
let container = apiSection.closest('div, p, fieldset');
while (container && container.parentElement) {
const reset = Array.from(container.querySelectorAll('a, button')).find(el =>
el.textContent.trim().toLowerCase() === 'reset' && el.offsetParent !== null
);
if (reset) { reset.click(); return true; }
container = container.parentElement;
if (container.tagName === 'FIELDSET' || container.tagName === 'FORM') break;
}
return false;
}""")
if reset_clicked:
page.wait_for_timeout(3000)
confirm = page.locator(
'button:has-text("OK"), button:has-text("Confirm"), button:has-text("Yes")'
)
if confirm.count() > 0 and confirm.first.is_visible():
confirm.first.click()
page.wait_for_timeout(3000)
api_el = page.query_selector("#api-access-key")
if api_el:
matches = re.findall(r"[a-f0-9]{40}", api_el.text_content())
if matches:
api_key = matches[0]
if not api_key:
log.error(f"[{name}] Could not get Redmine API key")
_debug_dump(page, f"redmine-no-key-{name}")
return ""
log.info(f"[{name}] Redmine API key obtained: {api_key[:8]}...")
bw.create_item(
name=item_name,
username=username,
password=api_key,
uris=[url],
collection_name=name,
)
log.info(f"[{name}] Redmine API key stored in Bitwarden")
return api_key
# ---------------------------------------------------------------------------
# Verification
# ---------------------------------------------------------------------------
def verify_gitea(token: str, agent: dict) -> bool:
"""Verify the Gitea API token works."""
import urllib.request
url = agent.get("systems", {}).get("gitea", {}).get("url", GITEA_URL)
req = urllib.request.Request(f"{url}/api/v1/user", headers={"Authorization": f"token {token}"})
try:
resp = urllib.request.urlopen(req, timeout=10)
data = json.loads(resp.read())
log.info(f" Gitea verify: user={data.get('login', '?')}")
return resp.status == 200
except Exception as e:
log.error(f" Gitea verify FAILED: {e}")
return False
def verify_redmine(key: str, agent: dict) -> bool:
"""Verify the Redmine API key works."""
import urllib.request
url = agent.get("systems", {}).get("redmine", {}).get("url", REDMINE_URL)
req = urllib.request.Request(f"{url}/users/current.json", headers={"X-Redmine-API-Key": key})
try:
resp = urllib.request.urlopen(req, timeout=10)
data = json.loads(resp.read())
log.info(f" Redmine verify: user={data.get('user', {}).get('login', '?')}")
return resp.status == 200
except Exception as e:
log.error(f" Redmine verify FAILED: {e}")
return False
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def load_manifest(path: str) -> list[dict]:
"""Load and validate the agent manifest."""
with open(path) as f:
data = yaml.safe_load(f)
agents = data.get("agents", [])
if not agents:
log.error("No agents found in manifest")
sys.exit(1)
for agent in agents:
if "cloudron_invite" not in agent or "REPLACE" in agent["cloudron_invite"]:
log.warning(f"Agent {agent.get('name', '?')} has no valid Cloudron invite")
return agents
def provision_agent(
context: BrowserContext,
agent: dict,
bw: BitwardenHelper,
phase1_only: bool = False,
) -> dict:
"""Provision a single agent identity."""
name = agent["name"]
results = {"name": name, "cloudron": False, "gitea": False, "discourse": False, "redmine": False}
page = context.new_page()
try:
# Phase 1: Cloudron enrollment
enroll_cloudron(page, agent, bw)
results["cloudron"] = True
log.info(f"[{name}] Phase 1 complete: Cloudron identity enrolled")
if phase1_only:
log.info(f"[{name}] Phase 1 only — skipping system access")
return results
systems = agent.get("systems", {})
if not systems:
log.info(f"[{name}] No systems configured — Phase 1 only")
return results
# Establish Cloudron panel session for SSO (shared across all apps)
sso_page = context.new_page()
cloudron_panel_login(sso_page, agent, bw)
sso_page.close()
# Phase 2: System access (fresh page for each system)
for system_name in ["gitea", "discourse", "redmine"]:
system_page = context.new_page()
try:
if system_name == "gitea":
token = provision_gitea(system_page, agent, bw)
results["gitea"] = bool(token) and verify_gitea(token, agent)
elif system_name == "discourse":
key = provision_discourse(system_page, agent, bw)
results["discourse"] = bool(key)
elif system_name == "redmine":
key = provision_redmine(system_page, agent, bw)
results["redmine"] = bool(key) and verify_redmine(key, agent)
except Exception as e:
log.error(f"[{name}] {system_name} provisioning failed: {e}")
finally:
system_page.close()
log.info(f"[{name}] All phases complete: {results}")
except Exception as e:
log.error(f"[{name}] Provisioning failed: {e}")
results["error"] = str(e)
finally:
page.close()
# Save state (in finally so partial results survive failures)
state_file = STATE_DIR / f"{name}.json"
try:
with open(state_file, "w") as f:
json.dump(results, f, indent=2)
except OSError:
log.warning(f"[{name}] Could not write state file: {state_file}")
return results
def main():
parser = argparse.ArgumentParser(description="Provision AI agent identities")
parser.add_argument("--manifest", default="agents.yaml", help="Path to manifest file")
parser.add_argument("--agent", help="Provision only this agent")
parser.add_argument("--phase1-only", action="store_true", help="Cloudron enrollment only")
parser.add_argument("--enable-2fa", action="store_true", help="Enable 2FA on existing Cloudron accounts")
parser.add_argument("--dry-run", action="store_true", help="Validate manifest without browser")
parser.add_argument("--headed", action="store_true", help="Show browser (debugging)")
args = parser.parse_args()
# Ensure state directory exists (deferred from module level so --dry-run works)
STATE_DIR.mkdir(parents=True, exist_ok=True)
# Load manifest
agents = load_manifest(args.manifest)
if args.agent:
agents = [a for a in agents if a["name"] == args.agent]
if not agents:
log.error(f"Agent '{args.agent}' not found in manifest")
sys.exit(1)
log.info(f"Manifest: {len(agents)} agent(s) to provision")
for a in agents:
log.info(f" - {a['name']} ({a.get('display_name', '?')}) [{a.get('priority', '?')}]")
if args.dry_run:
log.info("Dry run — manifest validated successfully")
return
# Initialize Bitwarden
bw = BitwardenHelper(
client_id=os.environ["BW_CLIENTID"],
client_secret=os.environ["BW_CLIENTSECRET"],
password=os.environ["BW_PASSWORD"],
totp_secret=os.environ.get("BW_TOTP_SECRET", ""),
server_url=os.environ.get("BW_SERVER", ""),
)
log.info("Connecting to Bitwarden...")
bw.login()
log.info("Bitwarden session established")
# Launch Playwright
headless = not args.headed
if os.environ.get("HEADFUL", "false").lower() == "true":
headless = False
all_results = []
with sync_playwright() as pw:
browser = pw.chromium.launch(headless=headless)
for agent in agents:
# Fresh context per agent (no cookie/session bleed)
context = browser.new_context(
accept_downloads=False,
java_script_enabled=True,
)
log.info(f"=== Provisioning: {agent['name']} ===")
try:
if args.enable_2fa:
page = context.new_page()
cloudron_panel_login(page, agent, bw)
totp_secret = enable_cloudron_2fa(page, agent, bw)
if totp_secret:
item_name = f"{agent['name']} Cloudron"
bw.update_item(item_name, totp_secret=totp_secret)
log.info(f"[{agent['name']}] Updated BW item with TOTP secret (in place)")
page.close()
all_results.append({"name": agent["name"], "2fa": bool(totp_secret)})
else:
result = provision_agent(context, agent, bw, args.phase1_only)
all_results.append(result)
except Exception as e:
log.error(f"FAILED: {agent['name']}: {e}")
all_results.append({"name": agent["name"], "error": str(e)})
finally:
context.close()
browser.close()
# Summary
log.info("\n=== PROVISIONING SUMMARY ===")
for r in all_results:
status = "OK" if "error" not in r else "FAILED"
systems = []
for s in ["cloudron", "gitea", "discourse", "redmine"]:
if r.get(s):
systems.append(s)
log.info(f" {r['name']}: {status} — {', '.join(systems) if systems else '(none)'}")
# Exit non-zero if any failed
if any("error" in r for r in all_results):
sys.exit(1)
if __name__ == "__main__":
main()
View Git Blame Copy Permalink