534 lines
17 KiB
Python
Executable File
534 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# /// script
|
|
# requires-python = ">=3.11"
|
|
# dependencies = [
|
|
# "garminconnect>=0.2.20",
|
|
# "curl_cffi>=0.7.0",
|
|
# ]
|
|
# ///
|
|
"""
|
|
Garmin Connect Authentication & Token Lifecycle Manager for OpenClaw.
|
|
|
|
Provides headless, 2-turn MFA login via Telegram, proactive token keepalive,
|
|
and token status inspection.
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional, Tuple
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
handlers=[logging.StreamHandler(sys.stderr)],
|
|
)
|
|
logger = logging.getLogger("garmin_auth")
|
|
|
|
|
|
def get_token_dir() -> str:
|
|
"""Get the token directory from environment or default locations."""
|
|
if os.environ.get("GARMINTOKENS"):
|
|
p = os.path.expanduser(os.environ["GARMINTOKENS"])
|
|
elif os.path.exists("/home/node/.openclaw"):
|
|
p = "/home/node/.openclaw/garmin_tokens"
|
|
else:
|
|
p = os.path.expanduser("~/.garminconnect")
|
|
os.makedirs(p, mode=0o700, exist_ok=True)
|
|
return p
|
|
|
|
|
|
def get_token_file() -> str:
|
|
return os.path.join(get_token_dir(), "garmin_tokens.json")
|
|
|
|
|
|
def get_token_base64_file() -> str:
|
|
return os.path.join(get_token_dir(), "garmin_tokens.base64")
|
|
|
|
|
|
def get_socket_path() -> str:
|
|
token_dir = get_token_dir()
|
|
if os.access(token_dir, os.W_OK):
|
|
return os.path.join(token_dir, "auth_daemon.sock")
|
|
return "/tmp/garmin_auth.sock"
|
|
|
|
|
|
def get_status_file() -> str:
|
|
token_dir = get_token_dir()
|
|
if os.access(token_dir, os.W_OK):
|
|
return os.path.join(token_dir, "auth_status.json")
|
|
return "/tmp/garmin_auth_status.json"
|
|
|
|
|
|
def get_credentials() -> Tuple[str, str]:
|
|
email = os.environ.get("GARMIN_EMAIL") or os.environ.get("garmin-email")
|
|
password = os.environ.get("GARMIN_PASSWORD") or os.environ.get("garmin-password")
|
|
|
|
if not email or not password:
|
|
# Check files if defined
|
|
email_file = os.environ.get("GARMIN_EMAIL_FILE")
|
|
if email_file and os.path.exists(email_file):
|
|
with open(email_file) as f:
|
|
email = f.read().strip()
|
|
password_file = os.environ.get("GARMIN_PASSWORD_FILE")
|
|
if password_file and os.path.exists(password_file):
|
|
with open(password_file) as f:
|
|
password = f.read().strip()
|
|
|
|
if not email or not password:
|
|
raise ValueError(
|
|
"GARMIN_EMAIL and GARMIN_PASSWORD environment variables must be set."
|
|
)
|
|
return email.strip(), password.strip()
|
|
|
|
|
|
def secure_tokens(token_dir: str) -> None:
|
|
"""Set secure 700 / 600 permissions."""
|
|
try:
|
|
if os.path.isdir(token_dir):
|
|
os.chmod(token_dir, 0o700)
|
|
for entry in os.scandir(token_dir):
|
|
if entry.is_file():
|
|
os.chmod(entry.path, 0o600)
|
|
except Exception as e:
|
|
logger.warning(f"Could not secure directory permissions: {e}")
|
|
|
|
|
|
def decode_jwt_exp(jwt_token: str) -> Tuple[Optional[datetime], Optional[int]]:
|
|
"""Extract expiration datetime and timestamp from JWT."""
|
|
try:
|
|
parts = jwt_token.split(".")
|
|
if len(parts) != 3:
|
|
return None, None
|
|
payload_b64 = parts[1]
|
|
payload_b64 += "=" * (-len(payload_b64) % 4)
|
|
payload = json.loads(base64.urlsafe_b64decode(payload_b64))
|
|
exp = payload.get("exp")
|
|
if exp:
|
|
dt = datetime.fromtimestamp(exp, tz=timezone.utc)
|
|
return dt, exp
|
|
except Exception:
|
|
pass
|
|
return None, None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI Command: status
|
|
# ---------------------------------------------------------------------------
|
|
def cmd_status() -> int:
|
|
token_file = get_token_file()
|
|
token_dir = get_token_dir()
|
|
status_file = get_status_file()
|
|
|
|
info: Dict[str, Any] = {
|
|
"token_dir": token_dir,
|
|
"token_file": token_file,
|
|
"exists": os.path.exists(token_file),
|
|
"valid": False,
|
|
"user": None,
|
|
"exp_utc": None,
|
|
"exp_timestamp": None,
|
|
"pending_mfa": False,
|
|
}
|
|
|
|
if os.path.exists(status_file):
|
|
try:
|
|
with open(status_file) as f:
|
|
s = json.load(f)
|
|
if s.get("state") == "waiting_for_mfa":
|
|
info["pending_mfa"] = True
|
|
info["mfa_started_at"] = s.get("started_at")
|
|
except Exception:
|
|
pass
|
|
|
|
if not info["exists"]:
|
|
print(json.dumps(info, indent=2))
|
|
return 1
|
|
|
|
try:
|
|
with open(token_file, "r") as f:
|
|
data = json.load(f)
|
|
|
|
di_token = data.get("di_token")
|
|
if di_token:
|
|
dt, exp = decode_jwt_exp(di_token)
|
|
if dt:
|
|
info["exp_utc"] = dt.isoformat()
|
|
info["exp_timestamp"] = exp
|
|
|
|
# Test live connectivity
|
|
from garminconnect import Garmin
|
|
|
|
client = Garmin()
|
|
client.login(token_dir)
|
|
full_name = client.get_full_name()
|
|
if full_name:
|
|
info["valid"] = True
|
|
info["user"] = full_name
|
|
except Exception as e:
|
|
info["error"] = str(e)
|
|
|
|
print(json.dumps(info, indent=2))
|
|
return 0 if info["valid"] else 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background Daemon: Run MFA Flow & Listen on Unix Socket
|
|
# ---------------------------------------------------------------------------
|
|
def _run_daemon():
|
|
"""Executed as a detached background worker process."""
|
|
socket_path = get_socket_path()
|
|
status_file = get_status_file()
|
|
token_dir = get_token_dir()
|
|
token_file = get_token_file()
|
|
base64_file = get_token_base64_file()
|
|
|
|
# Clean up old socket
|
|
if os.path.exists(socket_path):
|
|
try:
|
|
os.unlink(socket_path)
|
|
except OSError:
|
|
pass
|
|
|
|
try:
|
|
email, password = get_credentials()
|
|
except Exception as e:
|
|
with open(status_file, "w") as f:
|
|
json.dump({"state": "error", "error": str(e)}, f)
|
|
sys.exit(1)
|
|
|
|
with open(status_file, "w") as f:
|
|
json.dump(
|
|
{
|
|
"state": "initiating_login",
|
|
"started_at": time.time(),
|
|
"email": email,
|
|
},
|
|
f,
|
|
)
|
|
|
|
from garminconnect import (
|
|
Garmin,
|
|
GarminConnectAuthenticationError,
|
|
GarminConnectConnectionError,
|
|
GarminConnectTooManyRequestsError,
|
|
)
|
|
|
|
try:
|
|
garmin = Garmin(email=email, password=password, return_on_mfa=True)
|
|
status, _ = garmin.login()
|
|
|
|
if status != "needs_mfa":
|
|
# Logged in directly without MFA
|
|
garmin.client.dump(token_dir)
|
|
secure_tokens(token_dir)
|
|
with open(token_file, "r") as tf:
|
|
raw = tf.read()
|
|
with open(base64_file, "w") as bf:
|
|
bf.write(base64.b64encode(raw.encode()).decode())
|
|
os.chmod(base64_file, 0o600)
|
|
name = garmin.get_full_name()
|
|
with open(status_file, "w") as f:
|
|
json.dump({"state": "success", "user": name}, f)
|
|
sys.exit(0)
|
|
|
|
# Needs MFA - set up listening Unix socket
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.bind(socket_path)
|
|
os.chmod(socket_path, 0o600)
|
|
sock.listen(1)
|
|
sock.settimeout(300.0) # 5 minutes timeout for user to submit code
|
|
|
|
with open(status_file, "w") as f:
|
|
json.dump(
|
|
{
|
|
"state": "waiting_for_mfa",
|
|
"started_at": time.time(),
|
|
"email": email,
|
|
"socket": socket_path,
|
|
},
|
|
f,
|
|
)
|
|
|
|
while True:
|
|
try:
|
|
conn, _ = sock.accept()
|
|
except socket.timeout:
|
|
with open(status_file, "w") as f:
|
|
json.dump(
|
|
{"state": "timed_out", "error": "MFA wait timed out after 5 minutes"},
|
|
f,
|
|
)
|
|
break
|
|
|
|
with conn:
|
|
raw_data = conn.recv(1024).decode().strip()
|
|
if not raw_data:
|
|
continue
|
|
|
|
try:
|
|
payload = json.loads(raw_data)
|
|
code = payload.get("code", "").strip()
|
|
except Exception:
|
|
code = raw_data.strip()
|
|
|
|
code = re.sub(r"\D", "", code)
|
|
if len(code) != 6:
|
|
response = {
|
|
"status": "invalid_code",
|
|
"error": "Code must be 6 digits",
|
|
}
|
|
conn.sendall(json.dumps(response).encode())
|
|
continue
|
|
|
|
try:
|
|
garmin.resume_login(None, code)
|
|
garmin.client.dump(token_dir)
|
|
secure_tokens(token_dir)
|
|
|
|
# Export base64 string
|
|
with open(token_file, "r") as tf:
|
|
token_str = tf.read()
|
|
with open(base64_file, "w") as bf:
|
|
bf.write(base64.b64encode(token_str.encode()).decode())
|
|
os.chmod(base64_file, 0o600)
|
|
|
|
name = garmin.get_full_name() or "User"
|
|
response = {"status": "success", "user": name}
|
|
conn.sendall(json.dumps(response).encode())
|
|
|
|
with open(status_file, "w") as f:
|
|
json.dump(
|
|
{
|
|
"state": "success",
|
|
"user": name,
|
|
"finished_at": time.time(),
|
|
},
|
|
f,
|
|
)
|
|
break # Success! Clean exit
|
|
|
|
except GarminConnectAuthenticationError as ae:
|
|
response = {
|
|
"status": "auth_error",
|
|
"error": f"MFA verification failed: {ae}",
|
|
}
|
|
conn.sendall(json.dumps(response).encode())
|
|
except Exception as ex:
|
|
response = {
|
|
"status": "error",
|
|
"error": f"Unexpected error during resume_login: {ex}",
|
|
}
|
|
conn.sendall(json.dumps(response).encode())
|
|
|
|
except Exception as e:
|
|
with open(status_file, "w") as f:
|
|
json.dump({"state": "error", "error": str(e)}, f)
|
|
finally:
|
|
if os.path.exists(socket_path):
|
|
try:
|
|
os.unlink(socket_path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI Command: request-code
|
|
# ---------------------------------------------------------------------------
|
|
def cmd_request_code() -> int:
|
|
socket_path = get_socket_path()
|
|
status_file = get_status_file()
|
|
|
|
# If daemon is already running and waiting, check status
|
|
if os.path.exists(socket_path) and os.path.exists(status_file):
|
|
try:
|
|
with open(status_file) as f:
|
|
st = json.load(f)
|
|
if st.get("state") == "waiting_for_mfa":
|
|
print(
|
|
"MFA_ALREADY_ACTIVE: A pending login session is already waiting for your MFA code. "
|
|
"Reply with the 6-digit code or wait for timeout."
|
|
)
|
|
return 0
|
|
except Exception:
|
|
pass
|
|
|
|
# Clean up stale files
|
|
for p in (socket_path, status_file):
|
|
if os.path.exists(p):
|
|
try:
|
|
os.unlink(p)
|
|
except OSError:
|
|
pass
|
|
|
|
# Spawn daemon process via subprocess.Popen detached
|
|
script_path = os.path.abspath(__file__)
|
|
proc = subprocess.Popen(
|
|
[sys.executable, script_path, "_daemon"],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
start_new_session=True,
|
|
)
|
|
logger.info(f"Spawned Garmin auth daemon (PID: {proc.pid})")
|
|
|
|
# Poll status file up to 25 seconds
|
|
start_wait = time.time()
|
|
while time.time() - start_wait < 25.0:
|
|
time.sleep(0.5)
|
|
if os.path.exists(status_file):
|
|
try:
|
|
with open(status_file) as f:
|
|
data = json.load(f)
|
|
state = data.get("state")
|
|
if state == "waiting_for_mfa":
|
|
print(
|
|
"MFA_REQUIRED: Garmin has sent a 6-digit verification code to your email/SMS. "
|
|
"Please reply with the code."
|
|
)
|
|
return 0
|
|
elif state == "success":
|
|
print(
|
|
f"SUCCESS: Logged in directly as {data.get('user')} without MFA. Tokens saved."
|
|
)
|
|
return 0
|
|
elif state == "error":
|
|
print(f"ERROR: Login failed: {data.get('error')}", file=sys.stderr)
|
|
return 1
|
|
except Exception:
|
|
pass
|
|
|
|
print("ERROR: Timed out waiting for Garmin SSO to send MFA code.", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI Command: verify <code>
|
|
# ---------------------------------------------------------------------------
|
|
def cmd_verify(code: str) -> int:
|
|
code = re.sub(r"\D", "", code.strip())
|
|
if len(code) != 6:
|
|
print(f"ERROR: Code must be exactly 6 digits, got {code!r}", file=sys.stderr)
|
|
return 1
|
|
|
|
socket_path = get_socket_path()
|
|
if not os.path.exists(socket_path):
|
|
print(
|
|
"ERROR: No active Garmin auth session found. Run 'request-code' first.",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
try:
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
sock.connect(socket_path)
|
|
sock.settimeout(35.0)
|
|
|
|
sock.sendall(json.dumps({"code": code}).encode())
|
|
raw_res = sock.recv(2048).decode()
|
|
sock.close()
|
|
|
|
res = json.loads(raw_res)
|
|
if res.get("status") == "success":
|
|
user = res.get("user", "Moritz Graf")
|
|
print(f"SUCCESS: Authenticated as {user}! Tokens saved successfully.")
|
|
return 0
|
|
else:
|
|
print(f"ERROR: {res.get('error', 'Authentication failed')}", file=sys.stderr)
|
|
return 1
|
|
except Exception as e:
|
|
print(f"ERROR: Failed communicating with auth daemon: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI Command: keepalive
|
|
# ---------------------------------------------------------------------------
|
|
def cmd_keepalive() -> int:
|
|
token_dir = get_token_dir()
|
|
token_file = get_token_file()
|
|
|
|
if not os.path.exists(token_file):
|
|
print(f"ERROR: No tokens found at {token_file}", file=sys.stderr)
|
|
return 1
|
|
|
|
try:
|
|
from garminconnect import Garmin
|
|
|
|
client = Garmin()
|
|
client.login(token_dir)
|
|
# Calling an API endpoint triggers _token_expires_soon() and auto-refresh + dump
|
|
name = client.get_full_name()
|
|
if not name:
|
|
print("ERROR: Login returned empty profile name", file=sys.stderr)
|
|
return 1
|
|
|
|
with open(token_file) as f:
|
|
data = json.load(f)
|
|
dt, exp = decode_jwt_exp(data.get("di_token", ""))
|
|
|
|
exp_str = dt.strftime("%Y-%m-%d %H:%M:%S UTC") if dt else "unknown"
|
|
print(f"KEEPALIVE_OK: Token valid for {name}. Current DI token expires at: {exp_str}")
|
|
return 0
|
|
except Exception as e:
|
|
print(f"ERROR: Keepalive check failed: {e}", file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Main Router
|
|
# ---------------------------------------------------------------------------
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
description="Garmin Authentication & Token Lifecycle CLI for OpenClaw"
|
|
)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
|
|
# status
|
|
subparsers.add_parser("status", help="Check stored token status and validity")
|
|
|
|
# request-code
|
|
subparsers.add_parser(
|
|
"request-code", help="Initiate Garmin login and request an MFA code"
|
|
)
|
|
|
|
# verify <code>
|
|
verify_parser = subparsers.add_parser(
|
|
"verify", help="Submit MFA code to the active auth session"
|
|
)
|
|
verify_parser.add_argument("code", help="6-digit verification code")
|
|
|
|
# keepalive
|
|
subparsers.add_parser(
|
|
"keepalive", help="Ping Garmin API and proactively refresh token if needed"
|
|
)
|
|
|
|
# internal daemon
|
|
subparsers.add_parser("_daemon", help=argparse.SUPPRESS)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.command == "_daemon":
|
|
_run_daemon()
|
|
elif args.command == "status":
|
|
sys.exit(cmd_status())
|
|
elif args.command == "request-code":
|
|
sys.exit(cmd_request_code())
|
|
elif args.command == "verify":
|
|
sys.exit(cmd_verify(args.code))
|
|
elif args.command == "keepalive":
|
|
sys.exit(cmd_keepalive())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|