Add change password

app/auth.py

@@ -0,0 +1,182 @@
+# app/auth_core.py
+
+from contextlib import closing
+import bcrypt
+
+from db import get_conn #, create_user, verify_user, get_role_for_user
+
+
+# ---------------------------------------------------------------------------
+#  DB Initialisierung: `users`-Tabelle
+# ---------------------------------------------------------------------------
+
+# def init_auth_db():
+#     """
+#     Legt die users-Tabelle an.
+#     WICHTIG: password_hash wird als TEXT gespeichert, damit sowohl
+#     streamlit-authenticator als auch dein eigener bcrypt-Code kompatibel sind.
+#     """
+#     with closing(get_conn()) as conn, conn:
+#         conn.execute(
+#             """
+#             CREATE TABLE IF NOT EXISTS users (
+#                 id INTEGER PRIMARY KEY AUTOINCREMENT,
+#                 username TEXT UNIQUE NOT NULL,
+#                 email TEXT,
+#                 password_hash TEXT NOT NULL,
+#                 role TEXT NOT NULL DEFAULT 'user'
+#             )
+#             """
+#         )
+
+
+# ---------------------------------------------------------------------------
+#  Benutzer anlegen
+# ---------------------------------------------------------------------------
+
+def create_user(
+    username: str, 
+    password: str, 
+    role: str = "user",
+    email: str | None = None,
+    firstname: str | None = None,
+    lastname: str | None = None
+) -> bool:
+
+    """
+    Passwort wird als bcrypt-Hash (TEXT) gespeichert.
+    """
+    pw_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
+
+    try:
+        with closing(get_conn()) as conn, conn:
+            conn.execute(
+                "INSERT INTO users (username, email, password_hash, role, firstname, lastname) VALUES (?, ?, ?, ?, ?, ?)",
+                (username, email, pw_hash, role, firstname, lastname),
+            )
+        return True
+    except Exception:
+        return False
+
+
+# ---------------------------------------------------------------------------
+#  Benutzer überprüfen (z.B. für deine alte Streamlit-Login-Maske)
+# ---------------------------------------------------------------------------
+
+def verify_user(username: str, password: str):
+    """
+    Vergleicht eingegebenes Passwort mit dem gespeicherten bcrypt-Hash.
+    Rückgabe: (True, role) oder (False, None)
+    """
+    with closing(get_conn()) as conn:
+        row = conn.execute(
+            "SELECT password_hash, role FROM users WHERE username = ?",
+            (username,),
+        ).fetchone()
+
+    if not row:
+        return False, None
+
+    stored_hash, role = row
+
+    # stored_hash ist TEXT → zurück nach bytes
+    ok = bcrypt.checkpw(password.encode("utf-8"), stored_hash.encode("utf-8"))
+
+    return (ok, role) if ok else (False, None)
+
+
+# ---------------------------------------------------------------------------
+#  Rolle für einen Benutzer holen (für Streamlit-Inhalte)
+# ---------------------------------------------------------------------------
+
+def get_role_for_user(username: str) -> str:
+    with closing(get_conn()) as conn:
+        row = conn.execute(
+            "SELECT role FROM users WHERE username = ?",
+            (username,),
+        ).fetchone()
+    return row[0] if row else "user"
+
+
+# ---------------------------------------------------------------------------
+#  Credential-Lader für streamlit-authenticator
+# ---------------------------------------------------------------------------
+
+def load_credentials_from_db() -> dict:
+    """
+    Baut das credentials-Dict so:
+    {
+        "usernames": {
+            "hansi": {
+                "email": "hansi@example.com",
+                "name": "hansi",
+                "password": "$2b$12$...",
+            },
+            ...
+        }
+    }
+    streamlit-authenticator prüft dann Login + Cookies automatisch.
+    """
+    creds = {"usernames": {}}
+
+    with closing(get_conn()) as conn:
+        rows = conn.execute(
+            "SELECT username, email, password_hash FROM users"
+        ).fetchall()
+
+    for username, email, pw_hash in rows:
+        # pw_hash kommt als TEXT
+        creds["usernames"][username] = {
+            "name": username,
+            "email": email or f"{username}@example.local",
+            "password": pw_hash,  # TEXT → streamlit-authenticator-kompatibel
+        }
+
+    return creds
+
+# ---------------------------------------------------------------------------
+#  Muss das Passwort geändert werden (Admin-Vorgabe)?
+# ---------------------------------------------------------------------------
+
+def needs_password_change(username: str) -> bool:
+    """Prüft, ob der User ein neues Passwort setzen muss (new_pwd = 1)."""
+    with closing(get_conn()) as conn:
+        row = conn.execute(
+            "SELECT new_pwd FROM users WHERE username = ?",
+            (username,),
+        ).fetchone()
+    if not row:
+        return False
+    return row[0] == 1
+
+# ---------------------------------------------------------------------------
+#  Passwort ändern durch den Benutzer
+# ---------------------------------------------------------------------------
+
+def update_password(username: str, new_password: str, reset_flag: bool = True) -> bool:
+    """
+    Setzt ein neues Passwort für den User.
+    Wenn reset_flag=True: new_pwd wird auf 0 zurückgesetzt.
+    """
+    pw_hash = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
+
+    with closing(get_conn()) as conn, conn:
+        if reset_flag:
+            conn.execute(
+                """
+                UPDATE users
+                SET password_hash = ?, new_pwd = 0
+                WHERE username = ?
+                """,
+                (pw_hash, username),
+            )
+        else:
+            conn.execute(
+                """
+                UPDATE users
+                SET password_hash = ?
+                WHERE username = ?
+                """,
+                (pw_hash, username),
+            )
+    return True