diff --git a/bw_helper.py b/bw_helper.py index 80d08a2..5efaa9f 100644 --- a/bw_helper.py +++ b/bw_helper.py @@ -127,8 +127,16 @@ class BitwardenHelper: return self._run_bw(["generate", "-ulns", "--length", str(length)]) def get_totp(self, item_name: str) -> str: - """Get the current TOTP code for a Bitwarden item.""" - return self._run_bw(["get", "totp", item_name]) + """Get the current TOTP code for a Bitwarden item. + + Resolves via exact-name match first -- bw's search is fuzzy and + shared substrings (e.g. "coo" in every tsgstaff-coo-* username) + make name-based gets ambiguous. + """ + item_id = self.get_item_id(item_name) + if not item_id: + raise RuntimeError(f"Item '{item_name}' not found") + return self._run_bw(["get", "totp", item_id]) # ------------------------------------------------------------------- # Item ID resolution -- the safe way to reference items diff --git a/diagnose-vp-compliance.py b/diagnose-vp-compliance.py new file mode 100644 index 0000000..1d623b8 --- /dev/null +++ b/diagnose-vp-compliance.py @@ -0,0 +1,89 @@ +#!/usr/bin/env python3 +""" +diagnose-vp-compliance.py -- Reproduce the vp-compliance login with full +DOM dumps at every step to find where it diverges from the other agents. + +Usage: + docker compose run --rm --entrypoint python3 provision diagnose-vp-compliance.py +""" + +import os, sys, time +from pathlib import Path +from playwright.sync_api import sync_playwright + +sys.path.insert(0, os.path.dirname(__file__)) +from bw_helper import BitwardenHelper + +CLOUDRON_BASE = "https://my.knownelement.com" +STATE_DIR = Path("/app/state") +ITEM = "vp-compliance Cloudron" +EMAIL = "tsgstaff-coo-vpcompliance@turnsys.com" + + +def dump(page, label): + ts = time.strftime("%H%M%S") + page.screenshot(path=str(STATE_DIR / f"diag-vpc-{label}-{ts}.png"), full_page=True) + body = page.evaluate("() => document.body.innerText.substring(0, 300)") + els = page.evaluate("""() => Array.from(document.querySelectorAll( + 'input, [role="button"], button, .ui.message')).map(el => ({ + tag: el.tagName, id: el.id, type: el.type||'', + vis: el.offsetParent !== null, text: (el.textContent||'').trim().substring(0,50)}))""") + print(f"--- {label} ---") + print(f"URL: {page.url}") + print(f"BODY: {body[:200]}") + for e in els: + print(f" <{e['tag']}> id={e['id']} type={e['type']} vis={e['vis']} text={e['text']!r}") + + +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() + # Resolve by exact name to dodge the fuzzy-search ambiguity + item_id = bw.get_item_id(ITEM) + item = json.loads(bw._run_bw(["get", "item", item_id])) + password = item["login"]["password"] + + with sync_playwright() as p: + browser = p.chromium.launch(headless=True) + context = browser.new_context(viewport={"width": 1280, "height": 1024}) + page = context.new_page() + page.goto(f"{CLOUDRON_BASE}/login.html", wait_until="networkidle", timeout=20000) + page.wait_for_timeout(2000) + dump(page, "01-login-page") + + page.wait_for_selector("#inputPassword", timeout=15000) + page.click("#inputUsername") + page.keyboard.type(EMAIL) + page.click("#inputPassword") + page.keyboard.type(password) + dump(page, "02-filled") + + btn = page.locator('[role="button"]:has-text("Log in")') + print(f"login button count={btn.count()} visible={btn.first.is_visible() if btn.count() else '?'}") + print(f"login button classes: {btn.first.get_attribute('class') if btn.count() else '?'}") + btn.first.click() + page.wait_for_timeout(5000) + dump(page, "03-after-submit") + + totp = page.query_selector("#inputTotpToken") + print(f"TOTP field present={totp is not None} visible={totp.is_visible() if totp else False}") + if totp and totp.is_visible(): + code = bw._run_bw(["get", "totp", item_id]) + totp.click() + page.keyboard.type(code) + page.locator("#totpTokenSubmitButton").click() + page.wait_for_timeout(5000) + dump(page, "04-after-totp") + + browser.close() + + +import json # noqa: E402 + +if __name__ == "__main__": + main() diff --git a/docker-compose.yml b/docker-compose.yml index 07b1725..9609b25 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -23,3 +23,6 @@ services: - ./provision-redmine.py:/app/provision-redmine.py:ro - ./merge-invites.py:/app/merge-invites.py:ro - ./dump-invite-page.py:/app/dump-invite-page.py:ro + - ./validate-all-logins.py:/app/validate-all-logins.py:ro + - ./diagnose-vp-compliance.py:/app/diagnose-vp-compliance.py:ro + - ./validate-coo.py:/app/validate-coo.py:ro diff --git a/validate-all-logins.py b/validate-all-logins.py new file mode 100644 index 0000000..ce2c655 --- /dev/null +++ b/validate-all-logins.py @@ -0,0 +1,182 @@ +#!/usr/bin/env python3 +""" +validate-all-logins.py -- End-to-end phase-1 validation for all COO agents. + +For each agent in the manifest (excluding non-COO identities): + 1. Fresh Cloudron login: username + password + TOTP from BW + 2. Assert we land on an authenticated page + 3. If the agent has Gitea/Redmine/Discourse credentials in BW, verify + each API credential against its API + +Prints a PASS/FAIL matrix. Exits 1 if any login fails. + +Usage: + docker compose run --rm --entrypoint python3 provision validate-all-logins.py +""" + +import json +import os +import sys +import time +import urllib.request +from pathlib import Path + +import yaml +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") +GITEA_URL = "https://git.knownelement.com" +REDMINE_URL = "https://projects.knownelement.com" +DISCOURSE_URL = "https://community.turnsys.com" +STATE_DIR = Path("/app/state") + +# Identities that are NOT owned by the COO BW account +EXCLUDE = {"cco", "ctpo", "board-secretariat", "director-audit"} + + +def verify_gitea(token): + req = urllib.request.Request( + f"{GITEA_URL}/api/v1/user", + headers={"Authorization": f"token {token}"}, + ) + with urllib.request.urlopen(req, timeout=15) as r: + return json.load(r).get("login") + + +def verify_redmine(key): + req = urllib.request.Request( + f"{REDMINE_URL}/users/current.json", + headers={"X-Redmine-API-Key": key}, + ) + with urllib.request.urlopen(req, timeout=15) as r: + return json.load(r).get("user", {}).get("login") + + +def verify_discourse(key): + req = urllib.request.Request( + f"{DISCOURSE_URL}/notifications.json", + headers={"User-Api-Key": key}, + ) + with urllib.request.urlopen(req, timeout=15) as r: + r.read() + return "ok" + + +def _validate_agent(bw, context, agent, results): + """Validate one agent: fresh login + API credential checks.""" + name = agent["name"] + email = agent.get("cloudron_email", f"{name}@turnsys.com") + item = f"{name} Cloudron" + row = {"name": name, "cloudron": False, "totp": False, + "gitea": None, "redmine": None, "discourse": None, "err": ""} + + try: + if not bw.item_exists(item): + row["err"] = "no BW item" + results.append(row) + print(f"[{name:20s}] FAIL: no Cloudron item in BW") + return + + password = bw.get_item_password(item) + + page = context.new_page() + page.goto(f"{CLOUDRON_BASE}/login.html", + wait_until="networkidle", timeout=20000) + page.wait_for_timeout(1500) + 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(4000) + + totp_el = page.query_selector("#inputTotpToken") + if totp_el and totp_el.is_visible(): + row["totp"] = True # 2FA prompt appeared = 2FA is ON + code = bw.get_totp(item) + totp_el.click() + page.keyboard.type(code) + page.locator("#totpTokenSubmitButton").click() + page.wait_for_timeout(5000) + + authed = ("login" not in page.url.lower() + and "openid" not in page.url.lower() + and "setupaccount" not in page.url.lower()) + row["cloudron"] = authed + if not authed: + page.screenshot(path=str( + STATE_DIR / f"validate-fail-{name}-{time.strftime('%H%M%S')}.png")) + row["err"] = f"login stuck at {page.url[:60]}" + page.close() + except Exception as e: + row["err"] = str(e)[:80] + + # API credential checks (failures recorded, not fatal) + for sysname, verifier, bwitem in [ + ("gitea", verify_gitea, f"{name} Gitea"), + ("redmine", verify_redmine, f"{name} Redmine"), + ("discourse", verify_discourse, f"{name} Discourse"), + ]: + try: + if bw.item_exists(bwitem): + row[sysname] = verifier(bw.get_item_password(bwitem)) + else: + row[sysname] = "-" + except Exception as e: + row[sysname] = f"ERR {str(e)[:40]}" + + results.append(row) + g = row["gitea"] if row["gitea"] is not None else "?" + r = row["redmine"] if row["redmine"] is not None else "?" + d = row["discourse"] if row["discourse"] is not None else "?" + status = "PASS" if row["cloudron"] and row["totp"] else "FAIL" + print(f"[{name:20s}] {status} cloudron={row['cloudron']} " + f"2fa={row['totp']} gitea={g} redmine={r} discourse={d} " + f"{row['err']}") + + +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() + + manifest = yaml.safe_load(open("/app/agents.yaml")) + agents = [a for a in manifest["agents"] if a["name"] not in EXCLUDE] + + results = [] + with sync_playwright() as p: + browser = p.chromium.launch(headless=True) + context = browser.new_context(viewport={"width": 1280, "height": 1024}) + + for agent in agents: + # Fresh context per agent: a shared context carries the previous + # agent's session cookie and login.html never shows the form. + context = browser.new_context(viewport={"width": 1280, "height": 1024}) + try: + _validate_agent(bw, context, agent, results) + finally: + context.close() + + browser.close() + + print("\n=== VALIDATION MATRIX ===") + fails = 0 + for row in results: + ok = row["cloudron"] and row["totp"] + if not ok: + fails += 1 + print(f" {row['name']:20s} {'OK ' if ok else 'FAIL'} {row['err']}") + + sys.exit(1 if fails else 0) + + +if __name__ == "__main__": + main() diff --git a/validate-coo.py b/validate-coo.py new file mode 100644 index 0000000..6cf56f2 --- /dev/null +++ b/validate-coo.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +""" +validate-coo.py -- Targeted re-validation of the coo identity. + +Verifies the get_totp exact-name fix: fresh login with password + TOTP. + +Usage: + docker compose run --rm --entrypoint python3 provision validate-coo.py +""" + +import os +import sys + +from playwright.sync_api import sync_playwright + +sys.path.insert(0, os.path.dirname(__file__)) +from bw_helper import BitwardenHelper + +CLOUDRON_BASE = "https://my.knownelement.com" +ITEM = "coo Cloudron" +EMAIL = "tsgstaff-coo@turnsys.com" + + +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() + password = bw.get_item_password(ITEM) + + with sync_playwright() as p: + browser = p.chromium.launch(headless=True) + context = browser.new_context(viewport={"width": 1280, "height": 1024}) + page = context.new_page() + + page.goto(f"{CLOUDRON_BASE}/login.html", wait_until="domcontentloaded", timeout=30000) + page.wait_for_timeout(1500) + 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(4000) + + totp_el = page.query_selector("#inputTotpToken") + if totp_el and totp_el.is_visible(): + print("2FA prompt: YES") + code = bw.get_totp(ITEM) + print(f"TOTP resolved by exact name: {code}") + totp_el.click() + page.keyboard.type(code) + page.locator("#totpTokenSubmitButton").click() + page.wait_for_timeout(5000) + else: + print("2FA prompt: NO (problem)") + + authed = ("login" not in page.url.lower() + and "openid" not in page.url.lower() + and "setupaccount" not in page.url.lower()) + print(f"coo login: {'PASS' if authed else 'FAIL'} ({page.url})") + sys.exit(0 if authed else 1) + + +if __name__ == "__main__": + main()