File size: 6,033 Bytes
89d9642 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """Accounts and profile persistence.
Two changes from the app this replaces:
* Passwords are hashed with bcrypt. The previous version stored them in
plaintext and compared with `user["password"] == login_pw`.
* MongoDB is optional. If no connection string is configured the app runs in
a session-only demo mode instead of crashing on import, so it can be
developed and reviewed without a database.
Set MONGODB_URI as a Hugging Face Space repository secret -- never in a file.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
import bcrypt
import streamlit as st
from nutriweb.profile.model import UserProfile
# bcrypt hashes at most 72 bytes and raises on anything longer (it used to
# truncate silently). Truncating here is the standard handling and keeps long
# passphrases usable.
BCRYPT_MAX_BYTES = 72
def _encode(password: str) -> bytes:
return (password or "").encode("utf-8")[:BCRYPT_MAX_BYTES]
def _hash(password: str) -> str:
return bcrypt.hashpw(_encode(password), bcrypt.gensalt()).decode()
@dataclass
class AuthResult:
ok: bool
message: str = ""
profile: UserProfile | None = None
def _uri() -> str:
"""Read the connection string, environment first.
Order matters. Hugging Face Space secrets arrive as environment variables,
and a stale `.streamlit/secrets.toml` left in the working tree must never
be able to override an explicitly-set variable -- reading the file first
silently redirected connections to an old cluster, which is exactly the
kind of failure that is invisible until it matters.
"""
from_env = os.environ.get("MONGODB_URI", "").strip()
if from_env:
return from_env
try:
if "MONGODB_URI" in st.secrets:
return str(st.secrets["MONGODB_URI"])
except Exception:
pass # no secrets.toml at all is the normal local state
return ""
@st.cache_resource(show_spinner=False)
def _collections():
"""Return (users, history) collections, or (None, None) in demo mode."""
uri = _uri()
if not uri:
return None, None
try:
from pymongo import MongoClient
client = MongoClient(uri, serverSelectionTimeoutMS=5000)
client.admin.command("ping")
db_name = os.environ.get("MONGO_DB_NAME", "nutriweb_db")
try:
db_name = str(st.secrets.get("MONGO_DB_NAME", db_name))
except Exception:
pass
db = client[db_name]
return db["users"], db["history"]
except Exception as exc: # unreachable database must not break the app
st.warning(f"Could not reach the account database ({exc}). Running in demo mode.")
return None, None
def _verify(password: str, hashed: str) -> bool:
"""Check a password without raising on a missing or malformed hash.
bcrypt raises on an empty or unrecognised hash string, which would turn a
corrupted record into a crash rather than a failed login.
"""
if not hashed:
return False
try:
return bcrypt.checkpw(_encode(password), hashed.encode("utf-8"))
except (ValueError, TypeError):
return False
def demo_mode() -> bool:
users, _ = _collections()
return users is None
def _demo_store() -> dict:
return st.session_state.setdefault("_demo_users", {})
def register(user_id: str, password: str, profile: UserProfile) -> AuthResult:
user_id = (user_id or "").strip()
if not user_id or not password:
return AuthResult(False, "Pick a username and a password.")
if len(password) < 8:
return AuthResult(False, "Use at least 8 characters for the password.")
record = {"_id": user_id, "password_hash": _hash(password), **profile.to_dict()}
users, _ = _collections()
if users is None:
store = _demo_store()
if user_id in store:
return AuthResult(False, "That username is taken.")
store[user_id] = record
else:
from pymongo.errors import DuplicateKeyError
try:
users.insert_one(record)
except DuplicateKeyError:
return AuthResult(False, "That username is taken.")
profile.user_id = user_id
return AuthResult(True, "Account created.", profile)
def login(user_id: str, password: str) -> AuthResult:
users, _ = _collections()
record = _demo_store().get(user_id) if users is None else users.find_one({"_id": user_id})
if not record or not _verify(password, record.get("password_hash", "")):
# Deliberately vague: distinguishing the two leaks which accounts exist.
return AuthResult(False, "Incorrect username or password.")
return AuthResult(True, "", UserProfile.from_dict(record))
def save_profile(profile: UserProfile) -> None:
users, _ = _collections()
if users is None:
store = _demo_store()
if profile.user_id in store:
store[profile.user_id].update(profile.to_dict())
else:
users.update_one({"_id": profile.user_id}, {"$set": profile.to_dict()})
def log_view(user_id: str, product: dict, verdict_summary: str) -> None:
"""Record that a user looked at a product, for the history panel."""
_, history = _collections()
entry = {
"user_id": user_id,
"code": product.get("code"),
"product_name": product.get("product_name"),
"brands": product.get("brands"),
"health_score": product.get("health_score"),
"verdict": verdict_summary,
}
if history is None:
st.session_state.setdefault("_demo_history", []).insert(0, entry)
else:
from datetime import datetime, timezone
history.insert_one({**entry, "timestamp": datetime.now(timezone.utc)})
def recent_views(user_id: str, limit: int = 25) -> list[dict]:
_, history = _collections()
if history is None:
return st.session_state.get("_demo_history", [])[:limit]
return list(
history.find({"user_id": user_id}).sort("timestamp", -1).limit(limit)
)
|