YahBible / code /auth_sync.py
OhBeOneKeyNoBe's picture
BlackBox: server-side cross-device login fallback + fetch timeout
6a57074 verified
Raw History Blame Contribute Delete
3.85 kB
"""Publish/fetch the split-key credential halves.
PC side (has tokens): publish the HF half to the public HF dataset and the GitHub half
to the public GitHub repo. Client side (no token): fetch both halves from the public URLs
so login works when the PC is offline (yahbible_auth.login reconstructs + decrypts).
Public is safe: each half is one leg of an OTP split (useless alone) and the whole is
still AES-GCM-encrypted under the password. The GitHub repo is the "passwords" store in
the sense that its data is encrypted, not that the repo is access-restricted -- a
restricted repo could not be read token-free by an offline client.
"""
from __future__ import annotations
import base64
import json
import subprocess
import urllib.request
HF_REPO = "OhBeOneKeyNoBe/yahbible-auth" # dataset
GH_REPO = "OhBeOneKeyNoBe/yahbible-auth"
HF_BASE = "https://huggingface.co/datasets/%s/resolve/main" % HF_REPO
GH_RAW = "https://raw.githubusercontent.com/%s/main" % GH_REPO
def publish_hf(uhash: str, hf_record: dict, token: str | None = None):
from huggingface_hub import upload_file
data = json.dumps(hf_record, separators=(",", ":")).encode()
upload_file(path_or_fileobj=data, path_in_repo="auth/%s.json" % uhash,
repo_id=HF_REPO, repo_type="dataset", token=token,
commit_message="auth half (HF) for %s" % uhash)
def _gh(args: list) -> tuple:
p = subprocess.run(["gh", "api"] + args, capture_output=True, text=True)
return p.returncode, p.stdout, p.stderr
def publish_gh(uhash: str, gh_record: dict):
"""Write auth/<uhash>.json via the GitHub contents API (gh CLI holds the token)."""
path = "repos/%s/contents/auth/%s.json" % (GH_REPO, uhash)
content = base64.b64encode(json.dumps(gh_record, separators=(",", ":")).encode()).decode()
rc, out, _ = _gh([path, "--jq", ".sha"]) # existing sha (for update) if present
args = [path, "--method", "PUT",
"-f", "message=auth half (GitHub) for %s" % uhash,
"-f", "content=%s" % content]
if rc == 0 and out.strip() and out.strip() != "null":
args += ["-f", "sha=%s" % out.strip()]
rc2, out2, err2 = _gh(args)
if rc2 != 0:
raise RuntimeError("gh publish failed: " + (err2 or out2)[:200])
def publish(record: dict, hf_token: str | None = None):
"""Publish both halves of a signup()/change_password() record."""
publish_hf(record["uhash"], record["hf"], token=hf_token)
publish_gh(record["uhash"], record["gh"])
def _alias_uhash(ident: str) -> str:
import hashlib
return hashlib.sha256(("yahbible:user:" + ident.strip().lower()).encode()).hexdigest()[:32]
def publish_aliases(record: dict, aliases: list, hf_token: str | None = None):
"""Publish the SAME record under extra lookup filenames (username / email / handle)
so a client can find it by whatever identifier the user types. The record content
(and its internal uhash = the AES-GCM AAD) is unchanged; only the filename varies."""
seen = set()
for a in aliases:
if not a:
continue
uh = _alias_uhash(a)
if uh in seen:
continue
seen.add(uh)
publish_hf(uh, record["hf"], token=hf_token)
publish_gh(uh, record["gh"])
def _get(url: str, timeout: float = 30) -> dict | None:
try:
req = urllib.request.Request(url, headers={"User-Agent": "yahbible"})
with urllib.request.urlopen(req, timeout=timeout) as r:
return json.loads(r.read().decode())
except Exception:
return None
def fetch_halves(uhash: str, timeout: float = 30) -> tuple:
"""Client-side, token-free: fetch (hf_record, gh_record) from the public URLs."""
return (_get("%s/auth/%s.json" % (HF_BASE, uhash), timeout),
_get("%s/auth/%s.json" % (GH_RAW, uhash), timeout))