TSYS Group COO
2d01a9f962
Discourse SSO flow: Cloudron login -> click "Log In" -> click OpenID Connect button -> complete signup (enter username) -> logged in. User API key generated via Discourse RSA-based flow: 1. Generate RSA keypair, submit public key 2. Authorize request on Discourse 3. Capture encrypted payload from POST response 4. Decrypt with PKCS1v15 padding (Discourse uses this, not OAEP) 5. Parse JSON to extract the key field API key verified working: User-Api-Key header returns 30 topics from /latest.json. Key stored in Bitwarden as "vp-techops Discourse". Redmine SSO is blocked: Cloudron returns "You do not have access" -- the vp-techops user needs app access granted by Cloudron admin. Also added cryptography==44.0.1 to requirements for RSA operations.
300 lines
12 KiB
Python
300 lines
12 KiB
Python
#!/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()
|