# SPDX-FileCopyrightText: 2023-2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
"""Utilities for creating and configuring a Flask application instance."""
import json
import logging
import os
from pathlib import Path
from dotenv import load_dotenv
from flask import Flask, has_request_context, session
from flask_bootstrap import Bootstrap
from flask_ldap3_login import LDAP3LoginManager
from flask_login import (
LoginManager,
UserMixin,
current_user,
user_logged_in,
user_logged_out,
)
from werkzeug.middleware.proxy_fix import ProxyFix
from labfrog.common import initialize_common, resolve_config_name
from labfrog.login import add_user, remove_user
login_manager = LoginManager()
ldap_manager = LDAP3LoginManager()
SESSION_AUTH_USER_KEY = "labfrog_auth_user"
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logging.getLogger("pymongo").setLevel(logging.ERROR)
[docs]
def normalize_config_string(value):
if value is None:
return ""
text = str(value).strip()
if len(text) >= 2 and text[0] == text[-1] and text[0] in {'"', "'"}:
text = text[1:-1].strip()
return text
[docs]
def infer_username_from_user_id(user_id: str) -> str:
"""Infer a display username from persisted auth identifiers.
Flask-Login stores `User.get_id()` in the session. For LDAP users this is
commonly a DN like `uid=jdoe,ou=users,...`. Prefer concise username-like
values for UI display, and fall back to the raw identifier when no safe
parse is available.
"""
text = str(user_id).strip()
if not text:
return ""
head = text.split(",", 1)[0].strip()
if "=" in head:
key, value = head.split("=", 1)
if key.strip().lower() in {"uid", "cn", "mail", "email"}:
candidate = value.strip()
if candidate:
return candidate
return text
[docs]
def infer_helmholtz_username_from_session() -> str:
"""Infer a Helmholtz display username from stored OIDC profile fields."""
profile = session.get("oidc_auth_profile")
if not isinstance(profile, dict):
return ""
for key in ("preferred_username", "email", "name", "sub"):
value = profile.get(key)
if isinstance(value, str) and value.strip():
return value.strip()
return ""
[docs]
def normalize_auth_method(app):
raw_method = app.config.get("AUTH_METHOD")
if not raw_method:
raw_method = "none" if app.config.get("LOGIN_DISABLED", False) else "ldap"
auth_method = str(raw_method).lower().strip()
if auth_method not in {"none", "ldap", "helmholtz"}:
logging.warning(
"Unsupported AUTH_METHOD %s; falling back to 'ldap'", auth_method
)
auth_method = "ldap"
app.config["AUTH_METHOD"] = auth_method
app.config["LOGIN_DISABLED"] = auth_method == "none"
return auth_method
[docs]
def apply_testing_cookie_overrides(app):
"""Force browser-session settings that work for local test clients."""
if not app.config.get("TESTING"):
return
original_secure = app.config.get("SESSION_COOKIE_SECURE")
original_samesite = app.config.get("SESSION_COOKIE_SAMESITE")
app.config["SESSION_COOKIE_SECURE"] = False
app.config["SESSION_COOKIE_SAMESITE"] = "Lax"
if original_secure is not False or original_samesite != "Lax":
logging.info(
"TESTING enabled; overriding session cookies to "
"SESSION_COOKIE_SECURE=False and SESSION_COOKIE_SAMESITE='Lax'."
)
[docs]
def ensure_oidc_client_secrets(app):
secrets_path = (
app.config.get("OIDC_CLIENT_SECRETS") or "helmholtz_client_secrets.json"
)
if not Path(secrets_path).is_absolute():
secrets_path = os.path.join(app.instance_path, secrets_path)
app.config["OIDC_CLIENT_SECRETS"] = secrets_path
server_url = (
app.config.get("OIDC_SERVER_URL") or "https://login.helmholtz.de/oauth2"
).rstrip("/")
redirect_uri = (
app.config.get("OIDC_REDIRECT_URI") or "http://127.0.0.1:5000/authorize"
)
provider_config = {
"web": {
"issuer": server_url,
"auth_uri": f"{server_url}/auth",
"token_uri": f"{server_url}/token",
"token_endpoint_auth_method": app.config.get(
"OIDC_TOKEN_AUTH_METHOD", "client_secret_basic"
),
"userinfo_uri": f"{server_url}/userinfo",
"token_introspection_uri": f"{server_url}/introspect",
"token_revocation_uri": f"{server_url}/revoke",
"client_id": app.config.get("OIDC_CLIENT_ID", ""),
"client_secret": app.config.get("OIDC_CLIENT_SECRET", ""),
"redirect_uris": [redirect_uri],
}
}
Path(os.path.dirname(secrets_path)).mkdir(exist_ok=True, parents=True)
with Path(secrets_path).open("w", encoding="utf-8") as fp:
json.dump(provider_config, fp, indent=2)
logging.info("Wrote OIDC client secrets to %s", secrets_path)
return secrets_path
[docs]
def load_oidc_credentials(app):
cred_path = app.config.get("OIDC_CREDENTIALS_FILE") or os.environ.get(
"OIDC_CREDENTIALS_FILE"
)
if not cred_path:
return
if not Path(cred_path).is_absolute():
cred_path = os.path.join(app.instance_path, cred_path)
if not Path(cred_path).exists():
logging.warning(
"OIDC_CREDENTIALS_FILE set to %s but file is missing", cred_path
)
return
found = {}
with Path(cred_path).open("r", encoding="utf-8") as fp:
for line in fp:
stripped = line.strip()
if not stripped or stripped.startswith("#") or "=" not in stripped:
continue
key, value = stripped.split("=", 1)
if key.strip() in {"OIDC_CLIENT_ID", "OIDC_CLIENT_SECRET"}:
found[key.strip()] = value.strip()
if found.get("OIDC_CLIENT_ID"):
app.config["OIDC_CLIENT_ID"] = found["OIDC_CLIENT_ID"]
if found.get("OIDC_CLIENT_SECRET"):
app.config["OIDC_CLIENT_SECRET"] = found["OIDC_CLIENT_SECRET"]
[docs]
def create_oidc_client():
from flask_oidc import OpenIDConnect
return OpenIDConnect()
[docs]
def create_app(test_config=None):
load_dotenv(dotenv_path=Path(__file__).resolve().parents[1] / ".env", override=True)
app = Flask(
__name__,
instance_path=os.environ.get("LABFROG_INSTANCE_PATH"),
instance_relative_config=True,
)
app.wsgi_app = ProxyFix(
app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_port=1, x_prefix=1
)
if test_config is None:
config_name = resolve_config_name()
app.config["CONFIG_NAME"] = config_name
app.config.from_pyfile(config_name)
else:
app.config.from_mapping(test_config)
app.config.setdefault("CONFIG_NAME", "test-config")
if app.config.get("TESTING"):
os.environ.setdefault("LABFROG_TESTING", "1")
apply_testing_cookie_overrides(app)
app.config.setdefault("USE_CUSTOM_OPTIONS", False)
app.config.setdefault("USE_WIKI", False)
app.config.setdefault("DOCS_URL", "https://labfrog.readthedocs.io/en/latest/")
app.config.setdefault("DOCS_LABEL", "Documentation")
# Request-time cache defaults. Set NAVIGATION_CACHE_SECONDS=0 for strict
# freshness when immediate cross-worker visibility is preferred.
app.config.setdefault("DIAGNOSTIC_CACHE_SECONDS", 10.0)
app.config.setdefault("HEADER_DOC_CACHE_SECONDS", 10.0)
app.config.setdefault("KNOWN_LAYOUT_FIELDS_CACHE_SECONDS", 10.0)
app.config.setdefault("CAMPAIGN_LAYOUT_DOC_CACHE_SECONDS", 30.0)
app.config.setdefault("CAMPAIGN_LAYOUT_SYNC_SECONDS", 30.0)
app.config.setdefault("NAVIGATION_CACHE_SECONDS", 5.0)
app.config.setdefault("LABFROG_LOCAL_TIMEZONE", "Europe/Berlin")
auth_method = normalize_auth_method(app)
if not app.config.get("USE_FULL_CUSTOM", True):
logging.info(
"Setting MONGODB COLLECTION to DEFAULT since not using FULL CUSTOM"
)
app.config["MONGODB_COLLECTION_FOR_SETTINGS"] = app.config[
"DEFAULT_COLLECTION_FOR_SETTINGS"
]
login_manager.init_app(app)
if not app.config.get("LDAP_HOST"):
app.config.setdefault("LDAP_HOST", "localhost")
app.config.setdefault("LDAP_BASE_DN", "")
app.config.setdefault("LDAP_USER_DN", "")
app.config.setdefault("LDAP_GROUP_DN", "")
app.config.setdefault("LDAP_USER_RDN", "uid")
app.config.setdefault("LDAP_USER_LOGIN_ATTR", "uid")
if auth_method == "helmholtz":
app.config.setdefault("OIDC_SCOPES", "openid profile email")
if isinstance(app.config.get("OIDC_SCOPES"), (list, tuple)):
app.config["OIDC_SCOPES"] = " ".join(app.config["OIDC_SCOPES"])
app.config.setdefault(
"OIDC_ID_TOKEN_COOKIE_SECURE",
app.config.get("SESSION_COOKIE_SECURE", False),
)
app.config.setdefault("OIDC_INTROSPECTION_AUTH_METHOD", "client_secret_basic")
app.config.setdefault("OIDC_TOKEN_AUTH_METHOD", "client_secret_basic")
app.config.pop("OIDC_CALLBACK_ROUTE", None)
load_oidc_credentials(app)
ensure_oidc_client_secrets(app)
oidc_client = create_oidc_client()
oidc_client.init_app(app)
app.extensions["oidc"] = oidc_client
else:
oidc_client = None
ldap_manager.init_app(app)
class User(UserMixin):
def __init__(self, dn, username, data):
self.dn = dn
self.username = username
self.data = data
def __repr__(self):
return self.dn
def get_id(self):
return self.dn
app.extensions["labfrog_user_factory"] = User
app.extensions["labfrog_user_store"] = {}
app.extensions["labfrog_logged_in_users"] = {}
app.extensions["labfrog_startup_state"] = {"session_cleared": False}
def _mark_user_active(username):
if not isinstance(username, str) or not username:
return
logged_in_users = app.extensions.setdefault("labfrog_logged_in_users", {})
if username not in logged_in_users:
add_user(username)
@login_manager.user_loader
def load_user(user_id):
store = app.extensions["labfrog_user_store"]
user = store.get(user_id)
if user:
_mark_user_active(getattr(user, "username", None))
return user
if not has_request_context() or not isinstance(user_id, str) or not user_id:
return None
auth_method = str(app.config.get("AUTH_METHOD", "")).strip().lower()
cached_user = session.get(SESSION_AUTH_USER_KEY)
if isinstance(cached_user, dict):
cached_id = cached_user.get("id")
cached_username = cached_user.get("username")
if (
isinstance(cached_id, str)
and cached_id == user_id
and isinstance(cached_username, str)
and cached_username
):
resolved_username = cached_username
if auth_method == "helmholtz":
resolved_username = (
infer_helmholtz_username_from_session() or resolved_username
)
elif auth_method == "ldap" and cached_username == cached_id:
resolved_username = (
infer_username_from_user_id(cached_id) or resolved_username
)
user = app.extensions["labfrog_user_factory"](
cached_id, resolved_username, {}
)
store[cached_id] = user
_mark_user_active(resolved_username)
session[SESSION_AUTH_USER_KEY] = {
"id": cached_id,
"username": resolved_username,
}
return user
# Backward compatibility for sessions created before SESSION_AUTH_USER_KEY
# was populated: preserve auth continuity by rehydrating with `user_id`.
if auth_method == "ldap":
inferred_username = infer_username_from_user_id(user_id)
elif auth_method == "helmholtz":
inferred_username = infer_helmholtz_username_from_session() or user_id
else:
inferred_username = user_id
user = app.extensions["labfrog_user_factory"](
user_id, inferred_username or user_id, {}
)
store[user_id] = user
_mark_user_active(inferred_username or user_id)
session[SESSION_AUTH_USER_KEY] = {
"id": user_id,
"username": inferred_username or user_id,
}
return user
@ldap_manager.save_user
def persist_user(dn, username, data, memberships):
store = app.extensions["labfrog_user_store"]
user = store.get(dn)
if user:
return user
user = app.extensions["labfrog_user_factory"](dn, username, data)
store[dn] = user
return user
@user_logged_in.connect_via(app)
def track_user_login(sender, user):
add_user(user.username)
@user_logged_out.connect_via(app)
def track_user_logout(sender, user):
username = getattr(user, "username", None)
if username:
remove_user(username)
@app.context_processor
def inject_demo_usage():
if not app.config.get("DEMO_MODE"):
return {}
try:
from labfrog.db import get_db
from labfrog.helpers.demo import demo_user_doc_count, demo_user_limit
username = (
current_user.username
if current_user and current_user.is_authenticated
else "anonymous"
)
shots = get_db()["shots"]
return {
"demo_doc_count": demo_user_doc_count(shots, username),
"demo_doc_limit": demo_user_limit(),
}
except Exception as exc:
logging.warning("Failed to compute demo usage: %s", exc)
return {}
@app.context_processor
def inject_footer_links():
docs_url = normalize_config_string(app.config.get("DOCS_URL"))
docs_label = normalize_config_string(app.config.get("DOCS_LABEL"))
return {
"footer_docs_url": docs_url,
"footer_docs_label": docs_label or "Documentation",
}
Path(app.instance_path).mkdir(exist_ok=True, parents=True)
initialize_common(app)
Bootstrap(app)
from . import db
db.init_app(app)
from . import (
add_entry,
diagnostic_definitions,
diagnostic_options,
edit_entry,
field_selection,
login,
new_parameter,
offline_queue,
organize_sections,
parameter_options,
search_records,
shot_details,
site_routes,
)
if auth_method == "helmholtz":
login.enable_oidc_login(app, oidc_client)
app.register_blueprint(login.blueprint)
app.register_blueprint(site_routes.blueprint)
app.register_blueprint(shot_details.blueprint)
app.register_blueprint(add_entry.blueprint)
app.register_blueprint(edit_entry.blueprint)
app.register_blueprint(diagnostic_options.blueprint)
app.register_blueprint(diagnostic_definitions.blueprint)
app.register_blueprint(field_selection.blueprint)
app.register_blueprint(search_records.blueprint)
app.register_blueprint(organize_sections.blueprint)
app.register_blueprint(parameter_options.blueprint)
app.register_blueprint(new_parameter.blueprint)
app.register_blueprint(offline_queue.blueprint)
app.config.setdefault("KAFKA_ENABLED", False)
app.config.setdefault("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")
app.config.setdefault("KAFKA_TOPIC", "")
app.config.setdefault("KAFKA_GROUP_ID", "labfrog-consumer")
app.config.setdefault("KAFKA_KEYS", "")
app.config.setdefault("KAFKA_AUTO_OFFSET_RESET", "latest")
from labfrog.kafka_listener import start_kafka_listener
start_kafka_listener(app)
return app