Add Hugging Face OAuth gate

app.py

@@ -1,5 +1,9 @@
+import os
+import secrets
 from typing import cast
+from urllib.parse import urlencode
 
+import requests
 import streamlit as st
 
 from gateway_client import delete_profile, ingest_and_rewrite
@@ -49,6 +53,105 @@ try:
 except FileNotFoundError:
     pass
 
+HF_CLIENT_ID = os.getenv("HF_OAUTH_CLIENT_ID")
+HF_CLIENT_SECRET = os.getenv("HF_OAUTH_CLIENT_SECRET")
+HF_REDIRECT_URI = os.getenv(
+    "HF_OAUTH_REDIRECT_URI", "https://memverge-memmachine-playground.hf.space/"
+)
+HF_SCOPES = os.getenv("HF_OAUTH_SCOPES", "openid profile email")
+HF_AUTH_URL = "https://huggingface.co/oauth/authorize"
+HF_TOKEN_URL = "https://huggingface.co/oauth/token"
+HF_PROFILE_URL = "https://huggingface.co/api/whoami-v2"
+HF_OAUTH_READY = bool(HF_CLIENT_ID and HF_CLIENT_SECRET and HF_REDIRECT_URI)
+
+
+def build_authorize_url(state: str) -> str:
+    params = {
+        "client_id": HF_CLIENT_ID,
+        "response_type": "code",
+        "redirect_uri": HF_REDIRECT_URI,
+        "scope": HF_SCOPES,
+        "state": state,
+    }
+    return f"{HF_AUTH_URL}?{urlencode(params)}"
+
+
+def exchange_code_for_token(code: str) -> dict | None:
+    payload = {
+        "grant_type": "authorization_code",
+        "code": code,
+        "client_id": HF_CLIENT_ID,
+        "client_secret": HF_CLIENT_SECRET,
+        "redirect_uri": HF_REDIRECT_URI,
+    }
+    try:
+        resp = requests.post(HF_TOKEN_URL, data=payload, timeout=15)
+        resp.raise_for_status()
+        return resp.json()
+    except Exception as e:
+        st.error(f"Failed to exchange auth code: {e}")
+        return None
+
+
+def fetch_profile(access_token: str) -> dict | None:
+    try:
+        resp = requests.get(
+            HF_PROFILE_URL,
+            headers={"Authorization": f"Bearer {access_token}"},
+            timeout=10,
+        )
+        resp.raise_for_status()
+        return resp.json()
+    except Exception as e:
+        st.error(f"Failed to fetch Hugging Face profile: {e}")
+        return None
+
+
+def ensure_authenticated() -> bool:
+    if not HF_OAUTH_READY:
+        return True
+
+    if "hf_profile" in st.session_state:
+        return True
+
+    query_params = st.experimental_get_query_params()
+    if "code" in query_params:
+        code = query_params["code"][0]
+        returned_state = query_params.get("state", [""])[0]
+        expected_state = st.session_state.get("hf_oauth_state", "")
+        if expected_state and returned_state != expected_state:
+            st.error("Authentication error: state mismatch. Please try again.")
+            st.experimental_set_query_params()
+            return False
+
+        token_data = exchange_code_for_token(code)
+        if token_data and token_data.get("access_token"):
+            st.session_state["hf_token"] = token_data["access_token"]
+            profile = fetch_profile(token_data["access_token"])
+            if profile:
+                st.session_state["hf_profile"] = profile
+                st.experimental_set_query_params()
+                return True
+
+    state = secrets.token_urlsafe(16)
+    st.session_state["hf_oauth_state"] = state
+    login_url = build_authorize_url(state)
+
+    st.markdown("### 🔐 Authentication Required")
+    st.write(
+        "Sign in with your Hugging Face account to continue using the MemMachine playground."
+    )
+    st.markdown(
+        f'<a href="{login_url}" target="_self"><button style="padding:0.6rem 1.6rem;border-radius:999px;background:#ffbe5c;border:none;font-size:1rem;font-weight:600;">Sign in with Hugging Face</button></a>',
+        unsafe_allow_html=True,
+    )
+    st.info("After logging in, you will be redirected back automatically.")
+    return False
+
+
+if not ensure_authenticated():
+    st.stop()
+
 
 # ──────────────────────────────────────────────────────────────
 # Sidebar
@@ -142,6 +245,10 @@ def append_user_turn(msgs: list[dict], new_user_msg: str) -> list[dict]:
 # Title
 # ──────────────────────────────────────────────────────────────
 st.title("MemMachine Chatbot")
+if "hf_profile" in st.session_state:
+    user = st.session_state["hf_profile"]
+    full_name = user.get("name") or user.get("fullname") or user.get("hf_username")
+    st.caption(f"Signed in as {full_name}")
 
 
 # ──────────────────────────────────────────────────────────────