#!/usr/bin/env python3 """One-shot runner: grab agent Curogram session from CDP 9223, change the account password to the AMD password via SetPassword, verify with a fresh login, and (only on full success) update CUROGRAM_AGENT_PASSWORD in the amd-agent credentials.env (preserving other lines + mode 0600). Reads the OLD password / username / target from credentials.env. NO secrets are hardcoded and NO secrets are printed (cookies/passwords are redacted in all output). Safe to run; account-settings only. """ from __future__ import annotations import json import os import re import sys SKILL_SCRIPTS = "/home/claude/repos/openclaw/.claude/skills/curogram/scripts" CRED_PATH = "/home/claude/.config/amd-agent/credentials.env" sys.path.insert(0, SKILL_SCRIPTS) import cdp_cookies # noqa: E402 import change_password as cp # noqa: E402 # Password mutations (setPassword / changePassword) live on the authentication # microservice, NOT api-v2. Verified by schema probe: setPassword EXISTS on # authentication.curogram.com with Password! scalar args; api-v2 has neither # field. The session check (current-session) stays on api-v2. cp.GRAPHQL_URL = "https://authentication.curogram.com/graphql" def load_env_file(path: str) -> dict: out: dict[str, str] = {} with open(path, "r", encoding="utf-8") as fh: for line in fh: line = line.rstrip("\n") if not line or line.lstrip().startswith("#") or "=" not in line: continue k, v = line.split("=", 1) k = k.strip() v = v.strip() if len(v) >= 2 and v[0] == v[-1] and v[0] in ("'", '"'): v = v[1:-1] out[k] = v return out def redact(s: str) -> str: return f"<{len(s)} chars>" if s else "" def update_cred_file(path: str, key: str, new_value: str) -> None: with open(path, "r", encoding="utf-8") as fh: lines = fh.readlines() pat = re.compile(rf"^{re.escape(key)}=") found = False for i, line in enumerate(lines): if pat.match(line.lstrip()) or pat.match(line): lines[i] = f"{key}={new_value}\n" found = True break if not found: if lines and not lines[-1].endswith("\n"): lines[-1] += "\n" lines.append(f"{key}={new_value}\n") with open(path, "w", encoding="utf-8") as fh: fh.writelines(lines) os.chmod(path, 0o600) def main() -> int: env = load_env_file(CRED_PATH) username = env.get("CUROGRAM_AGENT_USERNAME", "") old_pw = env.get("CUROGRAM_AGENT_PASSWORD", "") new_pw = env.get("AMD_PASSWORD", "") print(json.dumps({ "step": "loaded_creds", "username": username, "old_pw": redact(old_pw), "new_pw_len": len(new_pw), })) if not (username and old_pw and new_pw): print(json.dumps({"step": "abort", "reason": "missing cred fields"})) return 2 # 1. Grab live session from the AGENT Chrome (CDP 9223). try: sess = cdp_cookies.get_cookies(9223, "curogram.com") except Exception as e: # noqa: BLE001 print(json.dumps({"step": "cdp", "ok": False, "error": str(e)[:300]})) return 2 cookie = sess["cookie_header"] xsrf = sess["xsrf_token"] # This deployment authenticates via the CurogramPracticeToken cookie; there # is no XSRF-TOKEN cookie. Cookie alone is sufficient (verified: GET # current-session returns 200 cookie-only). xsrf may be empty. print(json.dumps({ "step": "cdp", "ok": bool(cookie), "cookie": redact(cookie), "xsrf": redact(xsrf), "tab_url": sess.get("tab_url", ""), })) if not cookie: print(json.dumps({"step": "abort", "reason": "no session cookie"})) return 2 # 2. Confirm we are on the AGENT account before changing anything. s_status, s_resp = cp.verify_session(cookie, xsrf) # Surface identifying fields only (the agent's own account, PHI-free). ident = {} if isinstance(s_resp, dict): for k in ("email", "username", "accountId", "name", "firstName", "lastName", "id"): if k in s_resp: ident[k] = s_resp[k] # session may nest under "account" or "user" for nk in ("account", "user", "provider"): if isinstance(s_resp.get(nk), dict): for k in ("email", "username", "name", "firstName", "lastName", "id", "accountId"): if k in s_resp[nk]: ident[f"{nk}.{k}"] = s_resp[nk][k] print(json.dumps({"step": "session", "status": s_status, "identity": ident, "keys": list(s_resp.keys()) if isinstance(s_resp, dict) else None})) if s_status != 200: print(json.dumps({"step": "abort", "reason": "session check != 200"})) return 2 # The session payload carries name fields, not the login email. Guard on # the agent account's identity: displayName "Exult Agent" + provider type + # the Exult Healthcare practice. Refuse if this looks like any other account # (e.g. Gautam's personal Curogram). acct = s_resp.get("account", {}) if isinstance(s_resp, dict) else {} cur_practice = s_resp.get("currentPractice", {}) if isinstance(s_resp, dict) else {} display = (acct.get("displayName") or "").strip().lower() first = (acct.get("firstName") or "").strip().lower() last = (acct.get("lastName") or "").strip().lower() practice = (cur_practice.get("name") or "").strip().lower() is_agent = ( display == "exult agent" and first == "exult" and last == "agent" and "exult" in practice ) print(json.dumps({ "step": "account_guard", "ok": is_agent, "displayName": acct.get("displayName"), "practice": cur_practice.get("name"), "userType": acct.get("userType"), })) if not is_agent: print(json.dumps({ "step": "abort", "reason": "session is NOT the Exult Agent account; refusing to " "change password", })) return 2 # 3. Change the password (SetPassword). c_status, c_resp = cp.change_password(cookie, xsrf, old_pw, new_pw) accepted = c_status == 200 and "errors" not in c_resp print(json.dumps({"step": "change", "status": c_status, "accepted": accepted, "response": c_resp})) if not accepted: # CRITICAL: verbatim rejection, STOP. No silent retry. print(json.dumps({"step": "abort", "reason": "password REJECTED", "verbatim": c_resp})) return 1 # 4. Verify via a fresh, cookieless login with the NEW password. v_status, v_resp = cp.verify_login(username, new_pw, "PROVIDER") v_ok = v_status == 200 and "errors" not in v_resp # Redact any token values from the verify response before printing. safe_v = v_resp print(json.dumps({"step": "verify_login", "status": v_status, "ok": v_ok, "response_type": (list(safe_v.get("data", {}).get("login", {}).keys()) if isinstance(safe_v, dict) and isinstance(safe_v.get("data"), dict) else None), "errors": safe_v.get("errors") if isinstance(safe_v, dict) else None})) if not v_ok: print(json.dumps({"step": "abort", "reason": "re-login with new pw failed; " "NOT updating cred file"})) return 1 # 5. Only now: update the stored credential. update_cred_file(CRED_PATH, "CUROGRAM_AGENT_PASSWORD", new_pw) after = load_env_file(CRED_PATH) updated_ok = after.get("CUROGRAM_AGENT_PASSWORD") == new_pw mode = oct(os.stat(CRED_PATH).st_mode & 0o777) print(json.dumps({"step": "cred_update", "ok": updated_ok, "mode": mode})) print(json.dumps({"step": "DONE", "changed": True, "verified": v_ok, "cred_updated": updated_ok})) return 0 if __name__ == "__main__": sys.exit(main())