box/bin/muse-signin.py

#!/usr/bin/env python3
"""
Automated muse.ai sign-in with in-browser OTP approval.

Usage:
  signin.py --email <email> [--otp <code>]
  
If --otp is not provided, the script will:
1. Navigate through login flow up to OTP prompt
2. Print "APPROVAL_NEEDED: OTP required for [redacted]"
3. Exit with code 2

The operator then obtains the OTP (via chat with user) and re-runs:
  signin.py --email <email> --otp <code>

This keeps credentials out of the automation — the OTP is provided
transiently via the operator, never stored.

Exit codes:
  0: Success / already logged in
  2: APPROVAL_NEEDED (OTP prompt reached, awaiting code)
  3: NEEDS_HUMAN (multi-account selection needs manual choice)
  4: NEEDS_SIGNUP (account not found / registration required)
  1: General error
"""
import json, urllib.request, websocket, time, sys, argparse, importlib.util

CDP_PORT_DEFAULT = 9410


def _registry_port(node):
    try:
        path = "/home/super/Projects/NetVM/bin/netvm-registry.py"
        spec = importlib.util.spec_from_file_location("netvm_registry", path)
        mod = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(mod)
        return mod.port_for(node)
    except Exception:
        return None


def is_registered_in_accounts(email):
    path = "/home/super/Projects/NetVM/ACCOUNTS.md"
    try:
        with open(path) as f:
            for line in f:
                if line.startswith("|") and email.lower() in line.lower():
                    return True
    except Exception:
        pass
    return False


def get_page(port):
    cdp_url = "http://127.0.0.1:%s/json/list" % port
    with urllib.request.urlopen(cdp_url, timeout=5) as r:
        ts = json.load(r)
    pages = [t for t in ts if t.get('type') == 'page']
    if not pages:
        print("ERROR: No page found", file=sys.stderr)
        sys.exit(1)
    return pages[0]

def ev(ws, expr, await_p=False):
    ws.send(json.dumps({
        "id": 1, "method": "Runtime.evaluate",
        "params": {"expression": expr, "returnByValue": True, "awaitPromise": await_p}
    }))
    resp = json.loads(ws.recv())
    return resp.get('result', {}).get('result', {}).get('value')

def main():
    p = argparse.ArgumentParser()
    p.add_argument('--email', required=True)
    p.add_argument('--otp', default=None)
    p.add_argument('--node', default=None,
                   help='node name: CDP port comes from the NODES.md registry')
    p.add_argument('--cdp-port', default=None,
                   help='explicit CDP port (overrides --node lookup)')
    p.add_argument('--account-name', default=None,
                   help='display-name hint for Meta multi-account selection '
                        '(never the "+1" entry)')
    args = p.parse_args()

    if is_registered_in_accounts(args.email):
        print("Registry check: [redacted] is registered in ACCOUNTS.md")

    if args.cdp_port:
        port = args.cdp_port
    elif args.node:
        port = _registry_port(args.node) or CDP_PORT_DEFAULT
    else:
        port = CDP_PORT_DEFAULT

    page = get_page(port)
    ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=15)

    # Step 1: Check if already logged in or blocked at verification gate
    title = ev(ws, "document.title")
    body = ev(ws, "document.body.innerText.slice(0,500)")
    url = ev(ws, "location.href") or ""
    if "access/verification" in url or "confirm your age" in body.lower():
        ig_link = ev(ws, """(async()=>{
            try {
                const r = await fetch('/api/hatch/age-confirmation/linking-web-auth?account_type=instagram', {
                    headers: {'Accept': 'application/json'}
                });
                const j = await r.json();
                return j && j.url ? j.url : null;
            } catch(e) { return null; }
        })()""", True)
        if ig_link:
            print(f"LINK_INSTAGRAM_URL: {ig_link}")
        print(f"NEEDS_HUMAN: Age verification required for {args.email} at {url}. Client intervention required to confirm age or link Instagram/Facebook.", file=sys.stderr)
        ws.close()
        sys.exit(3)

    if ("Connected" in body or "Chats" in body) and "Enter your code" not in body and "access/verification" not in url:
        print(f"Already logged in (title: {title})")
        ws.close()
        return 0

    already_on_otp = ("Enter your code" in body or "code we sent" in body) and args.email in body
    if not already_on_otp:
        # Step 2: Click Log in (if on landing page)
        if "Log in" in body:
            print("Clicking Log in...")
            ev(ws, """(async()=>{
              const b=[...document.querySelectorAll('button')].find(x=>x.innerText.includes('Log in'));
              if(b) b.click(); return !!b;
            })()""", True)
            time.sleep(3)

        # Step 3: Enter email
        print("Entering email: [redacted]")
        result = ev(ws, f"""(async()=>{{
          const inp=[...document.querySelectorAll('input')].find(i=>
            (i.placeholder&&i.placeholder.toLowerCase().includes('email'))||
            (i.getAttribute('aria-label')&&i.getAttribute('aria-label').toLowerCase().includes('email'))
          );
          if(!inp) return 'NOINPUT';
          inp.focus();
          document.execCommand('insertText',false,'{args.email}');
          await new Promise(r=>setTimeout(r,500));
          return 'entered:'+inp.value;
        }})()""", True)
        print("Email: [redacted]")
        if result == 'NOINPUT':
            print("ERROR: Email input not found", file=sys.stderr)
            ws.close()
            sys.exit(1)
        time.sleep(1)

        # Step 4: Click Continue
        print("Clicking Continue...")
        ev(ws, """(async()=>{
          const b=[...document.querySelectorAll('button')].find(x=>x.innerText.includes('Continue'));
          if(b) b.click(); return !!b;
        })()""", True)
        time.sleep(4)
    else:
        print("Page is already on OTP prompt for this email, skipping email entry.")

    # Step 5: Check for OTP prompt
    body = ev(ws, "document.body.innerText.slice(0,500)")
    if "Enter your code" in body or "code we sent" in body or "To log in" in body or "To confirm your account" in body:
        print("OTP prompt detected for [redacted]")
        if not args.otp:
            print("APPROVAL_NEEDED: OTP required for [redacted]")
            print("Re-run with: signin.py --email [redacted] --otp <code>")
            ws.close()
            sys.exit(2)  # Approval needed
        
        # Enter OTP
        print("Entering OTP...")
        result = ev(ws, f"""(async()=>{{
          const inp=[...document.querySelectorAll('input')].find(i=>i.type==='text'||i.type==='number');
          if(!inp) return 'NOINPUT';
          inp.focus();
          document.execCommand('insertText',false,'{args.otp}');
          await new Promise(r=>setTimeout(r,500));
          return 'entered:'+inp.value;
        }})()""", True)
        print("OTP: [redacted]")
        time.sleep(1)
        
        # Click Next/Verify/Confirm
        print("Submitting OTP...")
        ev(ws, """(async()=>{
          const b=[...document.querySelectorAll('button')].find(x=>
            x.innerText.includes('Next')||x.innerText.includes('Verify')||x.innerText.includes('Confirm')
          );
          if(b) b.click(); return !!b;
        })()""", True)
        time.sleep(5)
        
        # Verify login
        body = ev(ws, "document.body.innerText.slice(0,500)")
        title = ev(ws, "document.title") or ""
        url = ev(ws, "location.href") or ""

        # Check for post-login gates requiring human/client intervention
        if "access/verification" in url or "confirm your age" in body.lower():
            ig_link = ev(ws, """(async()=>{
                try {
                    const r = await fetch('/api/hatch/age-confirmation/linking-web-auth?account_type=instagram', {
                        headers: {'Accept': 'application/json'}
                    });
                    const j = await r.json();
                    return j && j.url ? j.url : null;
                } catch(e) { return null; }
            })()""", True)
            if ig_link:
                print(f"LINK_INSTAGRAM_URL: {ig_link}")
            print(f"NEEDS_HUMAN: Age verification required for {args.email} at {url}. Client intervention required to confirm age or link Instagram/Facebook.", file=sys.stderr)
            ws.close()
            sys.exit(3)

        if "Connected" in body or "Chats" in body or (("Muse" in title or "Chat" in title) and "access/verification" not in url):
            print(f"SUCCESS: Logged in as [redacted] (title: {title})")
            ws.close()
            return 0
        # Multi-account selection: Meta shows one button per matching
        # account plus a "+1" collapsed entry. The "+1" is NOT a real
        # account — click the button whose text contains the hinted
        # display name. Without a hint (or no match), a human must pick.
        if "Your email matches multiple accounts" in ev(
                ws, "document.body.innerText.slice(0,2000)"):
            if not args.account_name:
                print("NEEDS_HUMAN: multiple accounts match and no "
                      "--account-name hint was given", file=sys.stderr)
                ws.close()
                sys.exit(3)
            picked = ev(ws, """(async()=>{
              const want=%s.toLowerCase();
              const btns=[...document.querySelectorAll('button')];
              const t=btns.find(b=>b.innerText.toLowerCase().includes(want)
                && !b.innerText.includes('+1'));
              if(t){t.click();return true;} return false;
            })()""" % json.dumps(args.account_name), True)
            if not picked:
                print(f"NEEDS_HUMAN: no account button matched "
                      f"'[redacted]'", file=sys.stderr)
                ws.close()
                sys.exit(3)
            print(f"Selected account '[redacted]', waiting...")
            time.sleep(5)
            body = ev(ws, "document.body.innerText.slice(0,500)")
            title = ev(ws, "document.title") or ""
            if "Connected" in body or "Chats" in body or "Muse" in title or "Chat" in title:
                print("SUCCESS: Logged in as [redacted] "
                      "(account: [redacted])")
                ws.close()
                return 0
            print("WARNING: account selection did not land in chat",
                  file=sys.stderr)
            ws.close()
            sys.exit(1)
        else:
            print(f"WARNING: Login may have failed. Title: {title}", file=sys.stderr)
            print(f"Body: {body[:150]}", file=sys.stderr)
            ws.close()
            sys.exit(1)
    else:
        err_msg = ev(ws, """(()=>{
          const el = document.querySelector('[role="alert"], .error, [data-error]');
          return el ? el.innerText.trim() : '';
        })()""")
        if err_msg:
            print(f"ERROR: {err_msg}", file=sys.stderr)
        else:
            print("No OTP prompt found - check page state", file=sys.stderr)
            print(f"Body: {body[:200]}", file=sys.stderr)
        ws.close()
        sys.exit(1)

if __name__ == '__main__':
    sys.exit(main())