#!/usr/bin/env python3 """ provision-discourse-apikey.py -- Generate Discourse User API key for vp-techops. Discourse User API Keys require an RSA-based flow: 1. Generate RSA key pair 2. Submit public key with the API key request 3. User authorizes the request 4. Decrypt the returned API key with private key Usage: docker compose run --rm --entrypoint python3 provision provision-discourse-apikey.py """ import base64 import json import os import re import secrets import sys import time import uuid from pathlib import Path from urllib.parse import quote_plus from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding, rsa from playwright.sync_api import sync_playwright sys.path.insert(0, os.path.dirname(__file__)) from bw_helper import BitwardenHelper CLOUDRON_BASE = os.environ.get("CLOUDRON_BASE", "https://my.knownelement.com") DISCOURSE_URL = os.environ.get("DISCOURSE_URL", "https://community.turnsys.com") STATE_DIR = Path("/app/state") BW_ITEM = "vp-techops Cloudron" DISCOURSE_BW_ITEM = "vp-techops Discourse" EMAIL = "tsgstaff-coo-vptechops@turnsys.com" USERNAME = "vptechops" def dump(page, label): ts = time.strftime("%H%M%S") try: page.screenshot(path=str(STATE_DIR / f"discourse-apikey-{label}-{ts}.png"), full_page=True) except Exception: pass body = page.evaluate("() => document.body.innerText.substring(0, 500)") print(f" [{label}] URL: {page.url}") print(f" Body: {body[:200]}") def cloudron_login(page, bw): password = bw.get_item_password(BW_ITEM) page.goto(f"{CLOUDRON_BASE}/login.html", wait_until="networkidle", timeout=15000) page.wait_for_timeout(2000) if "login" in page.url.lower() or "openid" in page.url.lower(): page.wait_for_selector("#inputPassword", timeout=15000) page.click("#inputUsername") page.keyboard.type(EMAIL) page.click("#inputPassword") page.keyboard.type(password) page.locator('[role="button"]:has-text("Log in")').first.click() page.wait_for_timeout(3000) totp = page.query_selector("#inputTotpToken") if totp and totp.is_visible(): code = bw.get_totp(BW_ITEM) totp.click() page.keyboard.type(code) page.locator("#totpTokenSubmitButton").click() page.wait_for_timeout(5000) def discourse_sso(page, bw): """Login to Discourse via SSO.""" page.goto(DISCOURSE_URL, wait_until="domcontentloaded", timeout=30000) page.wait_for_timeout(3000) if page.query_selector("#current-user, .current-user"): return True 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) 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 return page.query_selector("#current-user, .current-user") is not None def main(): bw = BitwardenHelper( client_id=os.environ["BW_CLIENTID"], client_secret=os.environ["BW_CLIENTSECRET"], password=os.environ["BW_PASSWORD"], server_url=os.environ.get("BW_SERVER", ""), ) bw.login() # Generate RSA key pair for User API Key flow private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) public_key = private_key.public_key() public_pem = public_key.public_bytes( encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo, ).decode("ascii") print(f"Generated RSA key pair (public key: {len(public_pem)} bytes)") nonce = secrets.token_hex(16) client_id = str(uuid.uuid4()) app_name = "TSG-Agent-VP-TechOps" with sync_playwright() as p: browser = p.chromium.launch(headless=True) context = browser.new_context(viewport={"width": 1280, "height": 1024}) page = context.new_page() # Step 1: Login print("=== STEP 1: Login ===") cloudron_login(page, bw) discourse_sso(page, bw) page.goto(DISCOURSE_URL, wait_until="domcontentloaded", timeout=30000) page.wait_for_timeout(3000) logged_in = page.query_selector("#current-user, .current-user") is not None print(f" Logged in: {logged_in}") if not logged_in: print(" FAILED to login") browser.close() return # Step 2: Request User API Key with RSA public key print("=== STEP 2: Request User API Key ===") 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)}" ) # Capture API responses api_responses = [] def handle_response(response): url = response.url if "user-api-key" in url and response.request.method == "POST": try: api_responses.append(response.text()) except Exception: pass page.on("response", handle_response) page.goto( f"{DISCOURSE_URL}/user-api-key/new{params}", wait_until="domcontentloaded", timeout=30000, ) page.wait_for_timeout(3000) dump(page, "01-apikey-request") body = page.evaluate("() => document.body.innerText") print(f" Page body: {body[:300]}") # Step 3: Authorize the request print("=== STEP 3: Authorize ===") authorized = False for btn_text in ["Authorize", "Approve", "Continue", "Allow", "Yes"]: loc = page.locator(f'button:has-text("{btn_text}"), [role="button"]:has-text("{btn_text}"), .btn-primary') if loc.count() > 0 and loc.first.is_visible(): loc.first.click() authorized = True page.wait_for_timeout(3000) print(f" Clicked: {btn_text}") break if not authorized: # Maybe it's a form with just a submit button submit = page.locator('button[type="submit"], input[type="submit"], .btn-primary') if submit.count() > 0 and submit.first.is_visible(): submit.first.click() authorized = True page.wait_for_timeout(3000) print(" Clicked submit button") dump(page, "02-after-authorize") # Step 4: Extract and decrypt API key print("=== STEP 4: Extract API key ===") api_key = "" # Check captured POST responses for resp_text in api_responses: print(f" Captured response: {resp_text[:200]}") try: data = json.loads(resp_text) encrypted_raw = data.get("key") or data.get("payload") or "" if encrypted_raw: encrypted_clean = encrypted_raw.replace("\n", "").replace("\r", "").replace(" ", "") encrypted_bytes = base64.b64decode(encrypted_clean) print(f" Encrypted payload: {len(encrypted_bytes)} bytes") # Try multiple padding schemes (Discourse version-dependent) paddings = [ ("OAEP-SHA256", padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None)), ("OAEP-SHA1", padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA1()), algorithm=hashes.SHA1(), label=None)), ("PKCS1v15", padding.PKCS1v15()), ] for name, pad in paddings: try: decrypted = private_key.decrypt(encrypted_bytes, pad).decode("ascii") # Decrypted payload is JSON: {"key":"...","nonce":"...","push":false,"api":4} try: key_data = json.loads(decrypted) api_key = key_data.get("key", decrypted) except json.JSONDecodeError: api_key = decrypted # fallback: key is plaintext print(f" Decrypted with {name}: {api_key[:12]}...") break except Exception: continue except Exception as e: print(f" Decryption failed: {e}") # If no captured response, check page body for JSON or plaintext key if not api_key: body = page.evaluate("() => document.body.innerText") # Try to find and decrypt the encrypted payload in the page # The page shows: "please paste the following key..." followed by base64 RSA-encrypted text key_match = re.search(r'(?:key|application):?\s*\n*\s*([A-Za-z0-9+/\n\r\s={30,}]+)', body) if key_match: encrypted = key_match.group(1).replace("\n", "").replace("\r", "").replace(" ", "").strip() try: api_key = private_key.decrypt( base64.b64decode(encrypted), padding.OAEP( mgf=padding.MGF1(algorithm=hashes.SHA256()), algorithm=hashes.SHA256(), label=None, ), ).decode("ascii") print(f" Decrypted from page body: {api_key[:12]}...") except Exception: pass # Look for unencrypted key (fallback) if not api_key: matches = re.findall(r'[a-f0-9]{64}', body) if matches: api_key = matches[0] print(f" Found unencrypted key: {api_key[:12]}...") if api_key: print(f"\n API KEY: {api_key}") print("=== STEP 5: Store in Bitwarden ===") existing = bw.get_item_id(DISCOURSE_BW_ITEM) if existing: bw.update_item(DISCOURSE_BW_ITEM, password=api_key) print(f" Updated BW item '{DISCOURSE_BW_ITEM}'") else: bw.create_item( name=DISCOURSE_BW_ITEM, username=USERNAME, password=api_key, uris=[DISCOURSE_URL], collection_name="default", ) print(f" Created BW item '{DISCOURSE_BW_ITEM}'") else: print(" Could not extract API key") # Dump all visible elements for debugging body = page.evaluate("() => document.body.innerText") print(f" Full body: {body[:500]}") browser.close() print("\n=== DONE ===") if __name__ == "__main__": main()