111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
import streamlit as st
|
|
import sqlite3
|
|
import bcrypt
|
|
from contextlib import closing
|
|
|
|
DB_PATH = "users.db"
|
|
|
|
# ---------- DB helpers ----------
|
|
def get_conn():
|
|
return sqlite3.connect(DB_PATH, check_same_thread=False)
|
|
|
|
def init_db():
|
|
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,
|
|
password_hash BLOB NOT NULL,
|
|
role TEXT NOT NULL DEFAULT 'user'
|
|
)
|
|
""")
|
|
|
|
def create_user(username: str, password: str, role: str = "user") -> bool:
|
|
pw_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
|
|
try:
|
|
with closing(get_conn()) as conn, conn:
|
|
conn.execute(
|
|
"INSERT INTO users (username, password_hash, role) VALUES (?, ?, ?)",
|
|
(username, pw_hash, role),
|
|
)
|
|
return True
|
|
except sqlite3.IntegrityError:
|
|
return False
|
|
|
|
def verify_user(username: str, password: str):
|
|
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
|
|
ok = bcrypt.checkpw(password.encode("utf-8"), stored_hash)
|
|
return ok, role if ok else (False, None)
|
|
|
|
# ---------- UI helpers ----------
|
|
def login_view():
|
|
st.title("Intranet Login")
|
|
with st.form("login"):
|
|
u = st.text_input("Username")
|
|
p = st.text_input("Passwort", type="password")
|
|
submitted = st.form_submit_button("Anmelden")
|
|
if submitted:
|
|
ok, role = verify_user(u.strip(), p)
|
|
if ok:
|
|
st.session_state["auth"] = True
|
|
st.session_state["username"] = u.strip()
|
|
st.session_state["role"] = role
|
|
st.success("Erfolgreich angemeldet.")
|
|
st.rerun()
|
|
else:
|
|
st.error("Login fehlgeschlagen.")
|
|
|
|
def authed_header():
|
|
st.sidebar.write(f"Angemeldet als **{st.session_state['username']}** ({st.session_state['role']})")
|
|
if st.sidebar.button("Logout"):
|
|
for k in ["auth", "username", "role"]:
|
|
st.session_state.pop(k, None)
|
|
st.rerun()
|
|
|
|
def content_for(role: str, username: str):
|
|
# Beispiel: rollen- und nutzerbezogene Inhalte
|
|
st.header("Dashboard")
|
|
st.info(f"Willkommen, {username}!")
|
|
|
|
if role == "admin":
|
|
st.subheader("Admin-Bereich")
|
|
st.write("Nur Admins sehen das hier.")
|
|
with st.expander("Neuen Nutzer anlegen"):
|
|
new_u = st.text_input("Neuer Username", key="new_u")
|
|
new_p = st.text_input("Neues Passwort", type="password", key="new_p")
|
|
new_role = st.selectbox("Rolle", ["user", "admin"], key="new_role")
|
|
if st.button("Anlegen"):
|
|
if new_u and new_p:
|
|
ok = create_user(new_u.strip(), new_p, new_role)
|
|
st.success("Nutzer angelegt.") if ok else st.error("Username bereits vorhanden.")
|
|
else:
|
|
st.warning("Bitte Username und Passwort eingeben.")
|
|
|
|
st.subheader("Dein Bereich")
|
|
st.write(f"Personalisierter Content für **{username}**.")
|
|
# Hier würdest du z. B. Daten aus einer Fach-DB nutzerbezogen filtern/anzeigen.
|
|
|
|
# ---------- App start ----------
|
|
def main():
|
|
st.set_page_config(page_title="Intranet-Portal", page_icon="🔒", layout="centered")
|
|
init_db()
|
|
|
|
# Optional: Initialer Admin (einmalig manuell auskommentieren, starten, dann wieder aus)
|
|
#create_user("admin", "change_me", role="admin")
|
|
|
|
if not st.session_state.get("auth"):
|
|
login_view()
|
|
else:
|
|
authed_header()
|
|
content_for(st.session_state["role"], st.session_state["username"])
|
|
|
|
if __name__ == "__main__":
|
|
main()
|