feat: add GPC SMS helper functionality with local SMS support
- Introduced GPC OTP channel selection (WhatsApp/SMS) in the sidepanel. - Added local SMS helper toggle and URL input for macOS users. - Implemented normalization functions for OTP channel and local SMS URL. - Updated settings payload to include new GPC helper configurations. - Enhanced Plus checkout process to utilize local SMS helper for OTP retrieval. - Added tests for GPC SMS helper script and its integration in the checkout flow. - Updated documentation to reflect new GPC helper features and configurations.
This commit is contained in:
QLHazyCoder
@@ -0,0 +1,313 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Local macOS SMS OTP helper for GPC GoPay flows.
|
||||
|
||||
Run this on the Mac that receives forwarded iPhone SMS messages. The helper
|
||||
reads a copy of the macOS Messages database and exposes the latest matching OTP
|
||||
on a localhost HTTP endpoint for the Chrome extension.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime as dt
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
import re
|
||||
import shutil
|
||||
import sqlite3
|
||||
import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
HOST = "127.0.0.1"
|
||||
PORT = 18767
|
||||
DEFAULT_DB = "~/Library/Messages/chat.db"
|
||||
MAC_ABSOLUTE_EPOCH = dt.datetime(2001, 1, 1, tzinfo=dt.timezone.utc)
|
||||
|
||||
OTP_PATTERNS = (
|
||||
re.compile(r"(?i)\bOTP\s*[:：]?\s*([0-9]{4,8})\b"),
|
||||
re.compile(r"#([0-9]{4,8})\b"),
|
||||
re.compile(r"(?<!\d)([0-9]{6})(?!\d)"),
|
||||
)
|
||||
KEYWORDS = ("gojek", "gopay", "openai llc", "openai")
|
||||
|
||||
STATE_LOCK = threading.Lock()
|
||||
STATE = {
|
||||
"started_at": dt.datetime.now(dt.timezone.utc).isoformat(),
|
||||
"db_path": "",
|
||||
"last_scan_at": "",
|
||||
"last_error": "",
|
||||
"last_rowid": 0,
|
||||
"last_otp": None,
|
||||
"otps": [],
|
||||
}
|
||||
|
||||
|
||||
def is_macos() -> bool:
|
||||
override = os.environ.get("GPC_SMS_HELPER_ALLOW_NON_MAC", "").strip().lower()
|
||||
return platform.system() == "Darwin" or override in {"1", "true", "yes"}
|
||||
|
||||
|
||||
def require_macos() -> None:
|
||||
if not is_macos():
|
||||
raise RuntimeError(
|
||||
"GPC 本地 SMS Helper 仅支持 macOS：需要读取 ~/Library/Messages/chat.db。"
|
||||
"请确认接收验证码的 iPhone 已开启短信转发，并能在 Mac 信息 app 中看到短信。"
|
||||
)
|
||||
|
||||
|
||||
def extract_gopay_otp(text: str, require_keywords: bool = True) -> Optional[str]:
|
||||
raw = text or ""
|
||||
lowered = raw.lower()
|
||||
if require_keywords and not any(keyword in lowered for keyword in KEYWORDS):
|
||||
return None
|
||||
for pattern in OTP_PATTERNS:
|
||||
match = pattern.search(raw)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return None
|
||||
|
||||
|
||||
def mac_message_time_to_datetime(value: int | float | None) -> dt.datetime:
|
||||
if not value:
|
||||
return dt.datetime.now(dt.timezone.utc)
|
||||
seconds = float(value)
|
||||
if seconds > 10_000_000_000:
|
||||
seconds = seconds / 1_000_000_000
|
||||
return MAC_ABSOLUTE_EPOCH + dt.timedelta(seconds=seconds)
|
||||
|
||||
|
||||
def update_state(**updates: object) -> None:
|
||||
with STATE_LOCK:
|
||||
STATE.update(updates)
|
||||
|
||||
|
||||
def get_state() -> dict:
|
||||
with STATE_LOCK:
|
||||
return dict(STATE)
|
||||
|
||||
|
||||
def copy_messages_db(db_path: Path) -> Path:
|
||||
if not db_path.exists():
|
||||
raise FileNotFoundError(f"Messages 数据库不存在：{db_path}")
|
||||
tmpdir = Path(tempfile.mkdtemp(prefix="gpc_messages_"))
|
||||
copied = tmpdir / "chat.db"
|
||||
shutil.copy2(db_path, copied)
|
||||
wal = db_path.with_name(db_path.name + "-wal")
|
||||
shm = db_path.with_name(db_path.name + "-shm")
|
||||
if wal.exists():
|
||||
shutil.copy2(wal, copied.with_name("chat.db-wal"))
|
||||
if shm.exists():
|
||||
shutil.copy2(shm, copied.with_name("chat.db-shm"))
|
||||
return copied
|
||||
|
||||
|
||||
def read_recent_messages(db_path: Path, after_rowid: int = 0, limit: int = 80) -> list[dict]:
|
||||
copied = copy_messages_db(db_path)
|
||||
try:
|
||||
conn = sqlite3.connect(str(copied))
|
||||
conn.row_factory = sqlite3.Row
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT
|
||||
message.ROWID AS rowid,
|
||||
message.guid AS guid,
|
||||
message.text AS text,
|
||||
message.date AS date,
|
||||
message.service AS service,
|
||||
handle.id AS handle
|
||||
FROM message
|
||||
LEFT JOIN handle ON message.handle_id = handle.ROWID
|
||||
WHERE message.ROWID > ?
|
||||
AND message.text IS NOT NULL
|
||||
ORDER BY message.ROWID DESC
|
||||
LIMIT ?
|
||||
""",
|
||||
(max(0, int(after_rowid or 0)), max(1, int(limit or 80))),
|
||||
).fetchall()
|
||||
return [dict(row) for row in rows]
|
||||
finally:
|
||||
try:
|
||||
conn.close() # type: ignore[name-defined]
|
||||
except Exception:
|
||||
pass
|
||||
shutil.rmtree(copied.parent, ignore_errors=True)
|
||||
|
||||
|
||||
def make_otp_record(row: dict, otp: str) -> dict:
|
||||
received_at = mac_message_time_to_datetime(row.get("date")).isoformat()
|
||||
return {
|
||||
"otp": otp,
|
||||
"code": otp,
|
||||
"message_id": str(row.get("guid") or row.get("rowid") or ""),
|
||||
"rowid": int(row.get("rowid") or 0),
|
||||
"sender": str(row.get("handle") or ""),
|
||||
"service": str(row.get("service") or ""),
|
||||
"received_at": received_at,
|
||||
"message_text": str(row.get("text") or ""),
|
||||
}
|
||||
|
||||
|
||||
def append_otp(record: dict, max_records: int = 30) -> None:
|
||||
with STATE_LOCK:
|
||||
records = [item for item in STATE.get("otps", []) if item.get("message_id") != record.get("message_id")]
|
||||
records.insert(0, record)
|
||||
STATE["otps"] = records[:max_records]
|
||||
STATE["last_otp"] = record
|
||||
STATE["last_rowid"] = max(int(STATE.get("last_rowid") or 0), int(record.get("rowid") or 0))
|
||||
|
||||
|
||||
def parse_timestamp_ms(value: object) -> int:
|
||||
if value is None:
|
||||
return 0
|
||||
if isinstance(value, (int, float)):
|
||||
numeric = float(value)
|
||||
else:
|
||||
raw = str(value).strip()
|
||||
if not raw:
|
||||
return 0
|
||||
try:
|
||||
numeric = float(raw)
|
||||
except ValueError:
|
||||
try:
|
||||
parsed = dt.datetime.fromisoformat(raw.replace("Z", "+00:00"))
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.replace(tzinfo=dt.timezone.utc)
|
||||
return int(parsed.timestamp() * 1000)
|
||||
except ValueError:
|
||||
return 0
|
||||
if numeric <= 0:
|
||||
return 0
|
||||
if numeric < 100_000_000_000:
|
||||
numeric *= 1000
|
||||
return int(numeric)
|
||||
|
||||
|
||||
def select_otp_record(state: dict, after_ms: int = 0) -> Optional[dict]:
|
||||
records = state.get("otps")
|
||||
if not isinstance(records, list):
|
||||
records = []
|
||||
if after_ms > 0:
|
||||
for record in records:
|
||||
if isinstance(record, dict) and parse_timestamp_ms(record.get("received_at")) >= after_ms:
|
||||
return record
|
||||
return None
|
||||
record = state.get("last_otp") or None
|
||||
if isinstance(record, dict):
|
||||
return record
|
||||
return records[0] if records and isinstance(records[0], dict) else None
|
||||
|
||||
|
||||
def scan_once(db_path: Path, require_keywords: bool = True) -> None:
|
||||
state = get_state()
|
||||
after_rowid = int(state.get("last_rowid") or 0)
|
||||
rows = read_recent_messages(db_path, after_rowid=after_rowid)
|
||||
max_rowid = after_rowid
|
||||
for row in reversed(rows):
|
||||
max_rowid = max(max_rowid, int(row.get("rowid") or 0))
|
||||
otp = extract_gopay_otp(row.get("text") or "", require_keywords=require_keywords)
|
||||
if otp:
|
||||
record = make_otp_record(row, otp)
|
||||
append_otp(record)
|
||||
print(f"captured OTP {otp} from message {record['message_id']} at {record['received_at']}", flush=True)
|
||||
update_state(last_rowid=max_rowid, last_scan_at=dt.datetime.now(dt.timezone.utc).isoformat(), last_error="")
|
||||
|
||||
|
||||
def scan_loop(db_path: Path, interval_seconds: float, require_keywords: bool) -> None:
|
||||
update_state(db_path=str(db_path))
|
||||
while True:
|
||||
try:
|
||||
scan_once(db_path, require_keywords=require_keywords)
|
||||
except Exception as exc:
|
||||
update_state(last_error=str(exc), last_scan_at=dt.datetime.now(dt.timezone.utc).isoformat())
|
||||
print(f"scan error: {exc}", file=sys.stderr, flush=True)
|
||||
time.sleep(max(0.5, float(interval_seconds or 2)))
|
||||
|
||||
|
||||
def write_json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> None:
|
||||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||||
handler.send_response(status)
|
||||
handler.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
handler.send_header("Access-Control-Allow-Origin", "*")
|
||||
handler.send_header("Cache-Control", "no-store")
|
||||
handler.send_header("Content-Length", str(len(body)))
|
||||
handler.end_headers()
|
||||
handler.wfile.write(body)
|
||||
|
||||
|
||||
class HelperHandler(BaseHTTPRequestHandler):
|
||||
server_version = "GpcSmsHelper/1.0"
|
||||
|
||||
def log_message(self, fmt: str, *args: object) -> None:
|
||||
print(f"{self.address_string()} - {fmt % args}", flush=True)
|
||||
|
||||
def do_GET(self) -> None: # noqa: N802
|
||||
parsed = urlparse(self.path)
|
||||
if parsed.path == "/health":
|
||||
state = get_state()
|
||||
write_json(self, 200, {
|
||||
"ok": True,
|
||||
"status": "ok",
|
||||
"has_otp": bool(state.get("last_otp")),
|
||||
"last_scan_at": state.get("last_scan_at"),
|
||||
"last_error": state.get("last_error"),
|
||||
})
|
||||
return
|
||||
if parsed.path in ("/otp", "/latest-otp"):
|
||||
query = parse_qs(parsed.query)
|
||||
consume = str(query.get("consume", ["0"])[0]).strip().lower() in {"1", "true", "yes"}
|
||||
after_ms = parse_timestamp_ms(query.get("after_ms", query.get("after", ["0"]))[0])
|
||||
state = get_state()
|
||||
record = select_otp_record(state, after_ms=after_ms)
|
||||
if not record:
|
||||
write_json(self, 200, {"ok": True, "otp": "", "code": "", "status": "waiting", "message": "未查询到验证码"})
|
||||
return
|
||||
payload = {"ok": True, "status": "found", **record}
|
||||
if consume:
|
||||
update_state(last_otp=None)
|
||||
write_json(self, 200, payload)
|
||||
return
|
||||
write_json(self, 404, {"ok": False, "error": "not_found"})
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="Run a local macOS Messages HTTP helper for GPC SMS OTP.")
|
||||
parser.add_argument("--host", default=HOST, help="Bind host, default 127.0.0.1")
|
||||
parser.add_argument("--port", type=int, default=PORT, help="Bind port, default 18767")
|
||||
parser.add_argument("--db", default=DEFAULT_DB, help="Path to macOS Messages chat.db")
|
||||
parser.add_argument("--interval", type=float, default=2.0, help="Message scan interval in seconds")
|
||||
parser.add_argument("--no-keywords", action="store_true", help="Accept any numeric OTP without GPC/OpenAI keywords")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
try:
|
||||
require_macos()
|
||||
db_path = Path(os.path.expanduser(args.db)).resolve()
|
||||
scanner = threading.Thread(
|
||||
target=scan_loop,
|
||||
args=(db_path, args.interval, not args.no_keywords),
|
||||
daemon=True,
|
||||
)
|
||||
scanner.start()
|
||||
server = ThreadingHTTPServer((args.host, int(args.port)), HelperHandler)
|
||||
print(f"GPC SMS Helper listening on http://{args.host}:{int(args.port)}", flush=True)
|
||||
print("请确认 iPhone 短信已转发到本机 Messages。", flush=True)
|
||||
server.serve_forever()
|
||||
return 0
|
||||
except KeyboardInterrupt:
|
||||
return 0
|
||||
except Exception as exc:
|
||||
print(str(exc), file=sys.stderr, flush=True)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user