moritz/mop-skills
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(skills): add garmin-auth skill with 2-turn Telegram MFA flow

Browse Source
This commit is contained in:
Moritz Graf
2026-09-17 09:29:56 +02:00
parent 0162f9dcd3
commit 2dfce6967c
2 changed files with 603 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/garmin-auth/SKILL.md
+70
View File
@@ -0,0 +1,70 @@
---
name: garmin-auth
description: Manage Garmin Connect authentication, proactive OAuth token keepalives, and interactive 2-turn MFA login over Telegram. Use when Garmin authentication fails (HTTP 401, expired tokens) or when the user asks to log into Garmin.
---
# 🔐 Garmin Connect Authentication & MFA Manager
Headless authentication lifecycle manager for Garmin Connect, allowing OpenClaw to initiate logins, keep tokens perpetually fresh, and verify 6-digit MFA codes directly via Telegram.
## 🎯 When to Use
- **Authentication Errors**: Whenever a `garmin` MCP tool call fails with `Garmin authentication failed`, `Tokens expired`, `401 Unauthorized`, or `Failed to retrieve social profile`.
- **User Request**: Whenever the user explicitly asks: *"Logge dich bei Garmin ein"*, *"Garmin MFA Code eingeben"*, *"Garmin Re-Auth"*, *"Verbinde Garmin"*.
- **Keepalive / Watchdog**: During morning briefings or background checks to ensure tokens are proactively refreshed before expiring.
- **Status Checks**: Whenever asked about Garmin token health or expiration: *"Ist Garmin eingeloggt?"*, *"Wie lange ist der Garmin-Token noch gültig?"*.
---
## 🕹️ Interactive Telegram MFA Login Flow (Protocol)
Because Garmin requires Multi-Factor Authentication (OTP sent via SMS/email), login operates in **two conversation turns**:
### Turn 1: Request Code (Initiate Login)
1. Execute the `request-code` command:
```bash
uv run /home/node/mop-skills/mop-skills/skills/garmin-auth/scripts/garmin_auth.py request-code
```
2. The command spawns a background authentication daemon and triggers Garmin to dispatch the 6-digit code to the user's email/SMS.
3. Immediately send a clear, friendly prompt to the user on Telegram:
> 🔑 **Garmin-Authentifizierung erforderlich**
> Ich habe einen neuen Login bei Garmin gestartet. Garmin hat dir soeben einen 6-stelligen Sicherheitscode per SMS oder E-Mail gesendet.
>
> Bitte antworte mir einfach kurz mit dem **6-stelligen Code**!
### Turn 2: Verify Code (Complete Login)
1. When the user sends a 6-digit message (e.g. `538291`, `538 291`):
2. Execute the `verify` command:
```bash
uv run /home/node/mop-skills/mop-skills/skills/garmin-auth/scripts/garmin_auth.py verify "<mfa_code>"
```
3. If successful:
- The tokens are automatically dumped to `/home/node/.openclaw/garmin_tokens/garmin_tokens.json` and base64 exported.
- Reply to the user:
> ✅ **Garmin-Login erfolgreich!**
> Angemeldet als **Moritz Graf**. Alle Garmin-Tools und Workouts sind wieder voll einsatzbereit.
4. If the code was incorrect:
- The daemon remains alive for 5 minutes. Inform the user and allow them to retry:
> ⚠️ Der eingegebene Code war leider ungültig oder abgelaufen. Bitte prüfe den Code und sende ihn erneut.
---
## 🛠️ CLI Operations Reference
### 1. Check Token Health & Validity
```bash
uv run /home/node/mop-skills/mop-skills/skills/garmin-auth/scripts/garmin_auth.py status
```
Outputs JSON with `valid`, `user`, `exp_utc`, `pending_mfa`.
### 2. Proactive Keepalive (Refresh Token before Expiry)
```bash
uv run /home/node/mop-skills/mop-skills/skills/garmin-auth/scripts/garmin_auth.py keepalive
```
Touches the API, triggering `python-garminconnect`'s automatic token refresh and persisting the newly rotated refresh token to `/home/node/.openclaw/garmin_tokens/garmin_tokens.json`.
---
## 🔒 Security & Environment
- Requires `GARMIN_EMAIL` and `GARMIN_PASSWORD` (already injected via `openclaw-secrets`).
- Persistent storage: `/home/node/.openclaw/garmin_tokens/garmin_tokens.json` (on PVC `openclaw-data`).
- Directory permissions are automatically set to `0700` and files to `0600`.
skills/garmin-auth/scripts/garmin_auth.py
Executable
+533
View File
@@ -0,0 +1,533 @@
#!/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()