# ============================================================ # MODULE: Authenticatie & Autorisatie — FASE 5.1 # HSEQ Intelligence Dashboard # Login, roles, multi-tenant bedrijfsisolatie # ============================================================ import os import sqlite3 from datetime import datetime from functools import wraps from flask import (jsonify, request, redirect, render_template_string, session, make_response) from werkzeug.security import generate_password_hash, check_password_hash from indexer import DB_PATH # ─── DB Init ─────────────────────────────────────────────────────────────── def init_db_auth(): conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS companies ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, sbi_code TEXT, brzo_classification TEXT, activities TEXT, created_at TEXT DEFAULT (datetime('now')) )''') c.execute('''CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT UNIQUE NOT NULL, password_hash TEXT NOT NULL, display_name TEXT, role TEXT NOT NULL DEFAULT 'readonly', company_id INTEGER, active INTEGER DEFAULT 1, created_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (company_id) REFERENCES companies(id) )''') c.execute('''CREATE TABLE IF NOT EXISTS notifications ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, company_id INTEGER, type TEXT NOT NULL, title TEXT NOT NULL, message TEXT, read INTEGER DEFAULT 0, created_at TEXT DEFAULT (datetime('now')), FOREIGN KEY (user_id) REFERENCES users(id), FOREIGN KEY (company_id) REFERENCES companies(id) )''') conn.commit() # Seed data row = c.execute("SELECT COUNT(*) FROM users WHERE email='admin'").fetchone() if row[0] == 0: c.execute("SELECT id FROM companies WHERE name='JvG Consultancy B.V.'") comp = c.fetchone() if not comp: c.execute("INSERT INTO companies (name,sbi_code,brzo_classification,activities) VALUES (?,?,?,?)", ("JvG Consultancy B.V.", "70220", "BRZO-inrichting (Seveso III, lageredrempel)", "HSEQ Consultancy, BRZO/Seveso compliance, veiligheidsrapportages")) comp_id = c.lastrowid else: comp_id = comp[0] c.execute("INSERT INTO users (email,password_hash,display_name,role,company_id) VALUES (?,?,?,?,?)", ("admin", generate_password_hash("Admin123!"), "Jorick van Gemert", "admin", comp_id)) conn.commit() conn.close() # ─── Helpers ─────────────────────────────────────────────────────────────── def get_current_user(): uid = session.get('user_id') if not uid: return None conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row user = conn.execute("SELECT u.*, c.name as company_name FROM users u LEFT JOIN companies c ON u.company_id=c.id WHERE u.id=? AND u.active=1", (uid,)).fetchone() conn.close() return user def require_login(f): @wraps(f) def decorated(*args, **kwargs): if not session.get('user_id'): return redirect(os.environ.get('BASE_PATH', '/hseq-dashboard') + '/login') return f(*args, **kwargs) return decorated def setup_auth_before_request(flask_app): """Register before_request hook to enforce login on all routes except login/static/api-auth.""" @flask_app.before_request def enforce_login(): from flask import request as req path = req.path bp = os.environ.get('BASE_PATH', '/hseq-dashboard') # Allow login page, login API, static files, and favicon # Check both with and without BASE_PATH prefix allowed = ['/login', '/api/auth/login', '/static', '/favicon.ico', bp + '/login', bp + '/api/auth/login', bp + '/static', bp + '/favicon.ico', bp + '/intelligence', bp + '/api/intelligence'] if any(path.startswith(p) for p in allowed): return None # Allow API auth endpoints (with and without prefix) if path.startswith('/api/auth/') or path.startswith(bp + '/api/auth/'): return None # Allow intelligence API endpoints for dashboard widgets if '/api/intelligence' in path: return None # Allow doc-scanner API endpoints (upload, browse, analyse, history, export) if '/doc-scanner/' in path: return None # Allow scraper status API if '/api/scraper/' in path: return None # Allow LMS score API (external quiz HTML posts here) if '/api/lms/score' in path: return None # Enforce login if not session.get('user_id'): return redirect(bp + '/login') return None def require_role(*roles): def decorator(f): @wraps(f) def decorated(*args, **kwargs): user = get_current_user() if not user: return redirect(os.environ.get('BASE_PATH', '/hseq-dashboard') + '/login') if user['role'] not in roles: return jsonify({"error": "Geen toegang"}), 403 return f(*args, **kwargs) return decorated return decorator def notification_count(): uid = session.get('user_id') if not uid: return 0 conn = sqlite3.connect(DB_PATH) cnt = conn.execute("SELECT COUNT(*) FROM notifications WHERE user_id=? AND read=0", (uid,)).fetchone()[0] conn.close() return cnt # ─── Login Page HTML ────────────────────────────────────────────────────── LOGIN_HTML = """ Login — HSEQ Intelligence Dashboard

HSEQ Intelligence Dashboard

JvG Consultancy

{% if error %}
{{error}}
{% endif %}
""" # ─── Admin Pages HTML ───────────────────────────────────────────────────── ADMIN_COMPANIES_HTML = """

🏢 Bedrijvenbeheer

Beheer bedrijfsprofielen voor multi-tenant isolatie.

{% for c in companies %} {% endfor %}
IDNaamSBIBRZOActiviteiten
{{c.id}} {{c.name}} {{c.sbi_code or '-'}} {{c.brzo_classification or '-'}} {{c.activities or '-'}}
""" ADMIN_USERS_HTML = """

👥 Gebruikersbeheer

Beheer gebruikersaccounts en roltoewijzingen.

{% for u in users %} {% endfor %}
IDE-mailNaamRolBedrijfActief
{{u.id}} {{u.email}} {{u.display_name or '-'}} {{u.role}} {{u.company_name or '-'}} {{'Ja' if u.active else 'Nee'}}
""" # ─── Register Routes ────────────────────────────────────────────────────── def register_auth_routes(flask_app, page_func, BASE_PATH="/hseq-dashboard"): app = flask_app # Ensure secret key if not app.secret_key: app.secret_key = os.environ.get('SECRET_KEY', 'hseq-intel-dev-secret-key-change-in-prod') # Fix session cookie path for nginx reverse proxy at /hseq-dashboard/ bp = os.environ.get('BASE_PATH', '/hseq-dashboard') app.config['SESSION_COOKIE_PATH'] = bp if bp else '/' app.config['APPLICATION_ROOT'] = bp if bp else None app.config['SESSION_COOKIE_SAMESITE'] = 'Lax' app.config['SESSION_COOKIE_SECURE'] = False # Enforce login on all routes setup_auth_before_request(app) @app.route(BASE_PATH + '/login') def login_page(): if session.get('user_id'): return redirect(bp + '/') return render_template_string(LOGIN_HTML, error=None) @app.route(BASE_PATH + '/api/auth/login', methods=['POST']) def auth_login(): email = request.form.get('email', '') password = request.form.get('password', '') remember = request.form.get('remember') conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row user = conn.execute("SELECT * FROM users WHERE email=? AND active=1", (email,)).fetchone() conn.close() if not user or not check_password_hash(user['password_hash'], password): return render_template_string(LOGIN_HTML, error="Ongeldig e-mailadres of wachtwoord.") session['user_id'] = user['id'] session['role'] = user['role'] session['company_id'] = user['company_id'] resp = make_response(redirect(bp + '/')) if remember: resp.set_cookie('remember_token', str(user['id']), max_age=30*24*3600, httponly=True) return resp @app.route(BASE_PATH + '/api/auth/logout', methods=['POST']) def auth_logout(): session.clear() resp = make_response(redirect(bp + '/login')) resp.delete_cookie('remember_token') return resp @app.route(BASE_PATH + '/api/auth/me') def auth_me(): user = get_current_user() if not user: return jsonify({"error": "Niet ingelogd"}), 401 return jsonify({ "id": user['id'], "email": user['email'], "display_name": user['display_name'], "role": user['role'], "company_name": user['company_name'] }) @app.route(BASE_PATH + '/admin/companies') @require_role('admin') def admin_companies(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row companies = conn.execute("SELECT * FROM companies ORDER BY name").fetchall() conn.close() return page_func(render_template_string(ADMIN_COMPANIES_HTML, companies=companies), active='admin-companies', page_title='Bedrijvenbeheer') @app.route(BASE_PATH + '/admin/companies', methods=['POST']) @require_role('admin') def admin_companies_create(): conn = sqlite3.connect(DB_PATH) conn.execute("INSERT INTO companies (name,sbi_code,brzo_classification,activities) VALUES (?,?,?,?)", (request.form['name'], request.form.get('sbi_code',''), request.form.get('brzo_classification',''), request.form.get('activities',''))) conn.commit() conn.close() return redirect(os.environ.get('BASE_PATH', '/hseq-dashboard') + '/admin/companies') @app.route(BASE_PATH + '/admin/companies/', methods=['PUT']) @require_role('admin') def admin_companies_update(cid): data = request.get_json() conn = sqlite3.connect(DB_PATH) conn.execute("UPDATE companies SET name=?, sbi_code=?, brzo_classification=?, activities=? WHERE id=?", (data.get('name',''), data.get('sbi_code',''), data.get('brzo_classification',''), data.get('activities',''), cid)) conn.commit() conn.close() return jsonify({"ok": True}) @app.route(BASE_PATH + '/admin/users') @require_role('admin') def admin_users(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row users = conn.execute("SELECT u.*, c.name as company_name FROM users u LEFT JOIN companies c ON u.company_id=c.id ORDER BY u.email").fetchall() companies = conn.execute("SELECT * FROM companies ORDER BY name").fetchall() conn.close() return page_func(render_template_string(ADMIN_USERS_HTML, users=users, companies=companies), active='admin-users', page_title='Gebruikersbeheer') @app.route(BASE_PATH + '/admin/users', methods=['POST']) @require_role('admin') def admin_users_create(): email = request.form['email'] password = request.form['password'] role = request.form.get('role', 'readonly') display_name = request.form.get('display_name', '') company_id = request.form.get('company_id') conn = sqlite3.connect(DB_PATH) try: conn.execute("INSERT INTO users (email,password_hash,display_name,role,company_id) VALUES (?,?,?,?,?)", (email, generate_password_hash(password), display_name, role, int(company_id) if company_id else None)) conn.commit() except Exception as e: conn.close() return f"Fout: {e}", 400 conn.close() return redirect(os.environ.get('BASE_PATH', '/hseq-dashboard') + '/admin/users')