diff --git a/application/single_app/app.py b/application/single_app/app.py index 8e0836d0..81e38efb 100644 --- a/application/single_app/app.py +++ b/application/single_app/app.py @@ -55,6 +55,7 @@ from route_frontend_feedback import * from route_frontend_support import * from route_frontend_notifications import * +from route_frontend_terms_of_use import register_route_frontend_terms_of_use from route_custom_pages import register_route_custom_pages from route_backend_chats import * @@ -98,6 +99,7 @@ from route_plugin_logging import bpl as plugin_logging_bp from functions_custom_pages import get_custom_pages_nav from functions_debug import debug_print +from functions_terms_of_use import has_terms_of_use_acceptance from opentelemetry.instrumentation.flask import FlaskInstrumentor @@ -813,6 +815,74 @@ def _is_idle_timeout_exempt(path): return any(path.startswith(prefix) for prefix in IDLE_TIMEOUT_EXEMPT_PREFIXES) +TERMS_OF_USE_EXEMPT_PATHS = { + '/login', + '/logout', + '/logout/local', + '/getAToken', + '/getATokenApi', + '/ci-auth/session', + '/auth/teams/token-exchange', + '/terms-of-use', + '/terms-of-use/accept', + '/terms-of-use/decline', + '/robots933456.txt', + '/favicon.ico', + '/acceptable_use_policy.html', + '/external/healthcheck', + '/external/healthcheckz', +} + +TERMS_OF_USE_EXEMPT_PREFIXES = ( + '/static/', + '/health', + '/api/health', +) + + +def _is_terms_of_use_exempt(path): + if path in TERMS_OF_USE_EXEMPT_PATHS: + return True + return any(path.startswith(prefix) for prefix in TERMS_OF_USE_EXEMPT_PREFIXES) + + +@app.before_request +def enforce_terms_of_use(): + """Block authenticated app usage until the current terms of use is accepted.""" + if 'user' not in session: + return None + if request.method == 'OPTIONS' or _is_terms_of_use_exempt(request.path): + return None + + request_settings = get_request_settings() + user_id = session.get('user', {}).get('oid') or session.get('user', {}).get('sub') + if has_terms_of_use_acceptance(request_settings, user_id=user_id): + return None + + terms_url = url_for( + 'frontend_terms_of_use.terms_of_use', + next=normalize_path_with_query(request.path, request.query_string), + ) + is_api_request = ( + request.accept_mimetypes.accept_json + and not request.accept_mimetypes.accept_html + ) or request.path.startswith('/api/') + if is_api_request: + return jsonify({ + 'error': 'terms_of_use_required', + 'message': 'Terms of Use acceptance is required before using SimpleChat.', + 'terms_url': terms_url, + }), 403 + return redirect(terms_url) + + +def normalize_path_with_query(path, query_string): + query_text = query_string.decode('utf-8', errors='ignore') if isinstance(query_string, bytes) else str(query_string or '') + if not query_text: + return path + return f"{path}?{query_text}" + + def maybe_log_authenticated_browser_request(): """Record throttled login activity for authenticated browser page requests.""" if request.method != 'GET' or request.path.startswith('/api/'): @@ -1079,6 +1149,9 @@ def list_semantic_kernel_plugins(): # ------------------- User Authentication Routes --------- register_route_blueprint('frontend_authentication', register_route_frontend_authentication) +# ------------------- Terms of Use Routes -- +register_route_blueprint('frontend_terms_of_use', register_route_frontend_terms_of_use) + # ------------------- User Profile Routes ---------------- register_route_blueprint('frontend_profile', register_route_frontend_profile, login_required_blueprint) diff --git a/application/single_app/config.py b/application/single_app/config.py index deadc08f..e8c8af0a 100644 --- a/application/single_app/config.py +++ b/application/single_app/config.py @@ -96,7 +96,7 @@ EXECUTOR_TYPE = 'thread' EXECUTOR_MAX_WORKERS = 30 SESSION_TYPE = 'filesystem' -VERSION = "0.250.052" +VERSION = "0.250.057" SESSION_COOKIE_SAMESITE = os.getenv('SESSION_COOKIE_SAMESITE', 'Lax') SESSION_COOKIE_HTTPONLY = os.getenv('SESSION_COOKIE_HTTPONLY', 'true').lower() != 'false' @@ -255,6 +255,9 @@ def get_allowed_extension_categories(enable_video=False, enable_audio=False): '/login', '/logout', '/logout/local', + '/terms-of-use', + '/terms-of-use/accept', + '/terms-of-use/decline', '/getAToken', '/getATokenApi', '/ci-auth/session', diff --git a/application/single_app/functions_activity_logging.py b/application/single_app/functions_activity_logging.py index fd88f1a9..10e7d5aa 100644 --- a/application/single_app/functions_activity_logging.py +++ b/application/single_app/functions_activity_logging.py @@ -1752,6 +1752,96 @@ def log_user_agreement_accepted( debug_print(f"⚠️ Warning: Failed to log user agreement acceptance: {str(e)}") +def log_terms_of_use_accepted( + user_id: str, + terms_hash: str, + frequency: str, + source: str, + accepted_date: str, + auth_state: Optional[str] = None, +) -> None: + """Log when a user accepts the terms of use.""" + try: + acceptance_record = { + 'id': str(uuid.uuid4()), + 'user_id': user_id, + 'activity_type': 'terms_of_use_accepted', + 'timestamp': datetime.utcnow().isoformat(), + 'created_at': datetime.utcnow().isoformat(), + 'accepted_date': accepted_date, + 'terms_hash': terms_hash, + 'frequency': frequency, + 'source': source, + 'auth_state': auth_state or 'authenticated', + } + + cosmos_activity_logs_container.create_item(body=acceptance_record) + log_event( + message=f"Terms of Use accepted: user {user_id}", + extra=acceptance_record, + level=logging.INFO, + ) + except Exception as e: + log_event( + message=f"Error logging terms of use acceptance: {str(e)}", + extra={ + 'user_id': user_id, + 'terms_hash': terms_hash, + 'frequency': frequency, + 'source': source, + 'error': str(e), + }, + level=logging.ERROR, + ) + debug_print(f"⚠️ Warning: Failed to log terms of use acceptance: {str(e)}") + + +def log_terms_of_use_declined( + user_id: str, + terms_hash: str, + frequency: str, + source: str, + redirect_url: str, + auth_state: Optional[str] = None, +) -> None: + """Log when a signed-in user declines the terms of use.""" + try: + decline_record = { + 'id': str(uuid.uuid4()), + 'user_id': user_id, + 'activity_type': 'terms_of_use_declined', + 'timestamp': datetime.utcnow().isoformat(), + 'created_at': datetime.utcnow().isoformat(), + 'declined_date': datetime.utcnow().strftime('%Y-%m-%d'), + 'terms_hash': terms_hash, + 'frequency': frequency, + 'source': source, + 'redirect_url': redirect_url, + 'auth_state': auth_state or 'authenticated', + } + + cosmos_activity_logs_container.create_item(body=decline_record) + log_event( + message=f"Terms of Use declined: user {user_id}", + extra=decline_record, + level=logging.INFO, + ) + except Exception as e: + log_event( + message=f"Error logging terms of use decline: {str(e)}", + extra={ + 'user_id': user_id, + 'terms_hash': terms_hash, + 'frequency': frequency, + 'source': source, + 'redirect_url': redirect_url, + 'error': str(e), + }, + level=logging.ERROR, + ) + debug_print(f"⚠️ Warning: Failed to log terms of use decline: {str(e)}") + + def has_user_accepted_agreement_today( user_id: str, workspace_type: str, diff --git a/application/single_app/functions_settings.py b/application/single_app/functions_settings.py index d48079f5..df351737 100644 --- a/application/single_app/functions_settings.py +++ b/application/single_app/functions_settings.py @@ -832,6 +832,13 @@ def get_settings(use_cosmos=False, include_source=False): 'logo_dark_version': 1, 'custom_favicon_base64': '', 'favicon_version': 1, + 'enable_terms_of_use': False, + 'terms_of_use_title': 'Terms of Use', + 'terms_of_use_message': '', + 'terms_of_use_frequency': 'once', + 'terms_of_use_decline_redirect_url': '/', + 'terms_of_use_accept_button_text': 'Accept and continue', + 'terms_of_use_decline_button_text': 'Cancel', 'enable_dark_mode_default': False, 'enable_left_nav_default': True, 'release_notifications_registered': False, diff --git a/application/single_app/functions_terms_of_use.py b/application/single_app/functions_terms_of_use.py new file mode 100644 index 00000000..134c1c65 --- /dev/null +++ b/application/single_app/functions_terms_of_use.py @@ -0,0 +1,307 @@ +# functions_terms_of_use.py +"""Helpers for the terms of use acceptance flow.""" + +import hashlib +import json +import logging +from datetime import datetime, timezone +from urllib.parse import urlparse + +from flask import has_request_context, session + +from functions_activity_logging import ( + log_terms_of_use_accepted, + log_terms_of_use_declined, +) +from functions_appinsights import log_event +from functions_settings import get_user_settings, update_user_settings + + +TERMS_OF_USE_FREQUENCIES = ("every_session", "daily", "once") +TERMS_OF_USE_DEFAULT_FREQUENCY = "once" +TERMS_OF_USE_SESSION_KEY = "terms_of_use_acceptance" +TERMS_OF_USE_PRE_AUTH_SESSION_KEY = "terms_of_use_pre_auth_acceptance" +TERMS_OF_USE_RETURN_PATH_SESSION_KEY = "terms_of_use_return_path" +TERMS_OF_USE_USER_SETTINGS_KEY = "termsOfUse" +TERMS_OF_USE_DEFAULT_REDIRECT = "/" +TERMS_OF_USE_MAX_TITLE_LENGTH = 160 +TERMS_OF_USE_MAX_MESSAGE_LENGTH = 10000 +TERMS_OF_USE_MAX_BUTTON_TEXT_LENGTH = 80 + + +def normalize_terms_of_use_frequency(value): + """Normalize the configured Terms of Use frequency.""" + normalized_value = str(value or "").strip().lower().replace("-", "_") + if normalized_value in {"session", "every_session", "per_session"}: + return "every_session" + if normalized_value in {"daily", "once_per_day", "per_day"}: + return "daily" + if normalized_value in {"once", "one_time", "just_once"}: + return "once" + return TERMS_OF_USE_DEFAULT_FREQUENCY + + +def normalize_terms_of_use_text(value, fallback="", max_length=TERMS_OF_USE_MAX_MESSAGE_LENGTH): + """Normalize administrator-entered Terms of Use text.""" + normalized_value = str(value or "").replace("\r\n", "\n").replace("\r", "\n").strip() + if not normalized_value: + normalized_value = fallback + return normalized_value[:max_length] + + +def normalize_terms_of_use_redirect_url(value): + """Return a safe local or admin-configured HTTP(S) redirect target.""" + candidate = str(value or "").strip() + if not candidate: + return TERMS_OF_USE_DEFAULT_REDIRECT + if "\\" in candidate: + return TERMS_OF_USE_DEFAULT_REDIRECT + if candidate.startswith("/") and not candidate.startswith("//"): + return candidate + + parsed = urlparse(candidate) + if ( + parsed.scheme == "https" + and parsed.netloc + and not parsed.username + and not parsed.password + ): + return candidate + return TERMS_OF_USE_DEFAULT_REDIRECT + + +def normalize_terms_of_use_return_path(value, fallback="/"): + """Normalize a return target so user-controlled redirects stay local.""" + candidate = str(value or "").strip() + if not candidate: + return fallback + if "\\" in candidate or not candidate.startswith("/") or candidate.startswith("//"): + return fallback + return candidate + + +def get_terms_of_use_config(settings): + """Build the normalized terms of use config from app settings.""" + source_settings = settings or {} + title = normalize_terms_of_use_text( + source_settings.get("terms_of_use_title"), + fallback="Terms of Use", + max_length=TERMS_OF_USE_MAX_TITLE_LENGTH, + ) + message = normalize_terms_of_use_text( + source_settings.get("terms_of_use_message"), + max_length=TERMS_OF_USE_MAX_MESSAGE_LENGTH, + ) + frequency = normalize_terms_of_use_frequency( + source_settings.get("terms_of_use_frequency") + ) + accept_button_text = normalize_terms_of_use_text( + source_settings.get("terms_of_use_accept_button_text"), + fallback="Accept and continue", + max_length=TERMS_OF_USE_MAX_BUTTON_TEXT_LENGTH, + ) + decline_button_text = normalize_terms_of_use_text( + source_settings.get("terms_of_use_decline_button_text"), + fallback="Cancel", + max_length=TERMS_OF_USE_MAX_BUTTON_TEXT_LENGTH, + ) + enabled = bool(source_settings.get("enable_terms_of_use", False) and message) + + return { + "enabled": enabled, + "title": title, + "message": message, + "frequency": frequency, + "decline_redirect_url": normalize_terms_of_use_redirect_url( + source_settings.get("terms_of_use_decline_redirect_url") + ), + "accept_button_text": accept_button_text, + "decline_button_text": decline_button_text, + "hash": compute_terms_of_use_hash(title, message, frequency), + } + + +def compute_terms_of_use_hash(title, message, frequency): + """Compute the hash used to invalidate old acceptances when terms change.""" + payload = { + "title": str(title or "").strip(), + "message": str(message or "").replace("\r\n", "\n").replace("\r", "\n").strip(), + "frequency": normalize_terms_of_use_frequency(frequency), + } + encoded_payload = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + return hashlib.sha256(encoded_payload).hexdigest() + + +def _utc_now(): + return datetime.now(timezone.utc) + + +def _build_acceptance_record(terms_config, accepted_at=None): + accepted_at = accepted_at or _utc_now() + return { + "hash": terms_config["hash"], + "frequency": terms_config["frequency"], + "accepted_at": accepted_at.isoformat(), + "accepted_date": accepted_at.strftime("%Y-%m-%d"), + } + + +def _session_acceptance_matches(terms_config): + if not has_request_context(): + return False + acceptance_record = session.get(TERMS_OF_USE_SESSION_KEY) + if not isinstance(acceptance_record, dict): + return False + return ( + acceptance_record.get("hash") == terms_config["hash"] + and acceptance_record.get("frequency") == terms_config["frequency"] + ) + + +def _user_acceptance_matches(terms_config, user_id): + if not user_id: + return False + + user_settings = get_user_settings(user_id) + terms_settings = (user_settings or {}).get("settings", {}).get(TERMS_OF_USE_USER_SETTINGS_KEY, {}) + if not isinstance(terms_settings, dict): + return False + if terms_settings.get("hash") != terms_config["hash"]: + return False + if terms_settings.get("frequency") != terms_config["frequency"]: + return False + + if terms_config["frequency"] == "once": + return True + if terms_config["frequency"] == "daily": + return terms_settings.get("accepted_date") == _utc_now().strftime("%Y-%m-%d") + return False + + +def has_terms_of_use_acceptance(settings, user_id=None): + """Return True when the user/session has satisfied the current Terms of Use.""" + terms_config = get_terms_of_use_config(settings) + if not terms_config["enabled"]: + return True + if not user_id and _session_acceptance_matches(terms_config): + return True + if terms_config["frequency"] == "every_session": + return _session_acceptance_matches(terms_config) + return _user_acceptance_matches(terms_config, user_id) + + +def mark_pre_auth_terms_of_use_acceptance(settings): + """Record anonymous pre-auth acceptance in the current Flask session.""" + terms_config = get_terms_of_use_config(settings) + if not terms_config["enabled"]: + return None + + acceptance_record = _build_acceptance_record(terms_config) + session[TERMS_OF_USE_PRE_AUTH_SESSION_KEY] = acceptance_record + session[TERMS_OF_USE_SESSION_KEY] = acceptance_record + session.modified = True + return acceptance_record + + +def record_terms_of_use_acceptance(user_id, settings, source="post_auth"): + """Persist and audit acceptance for an authenticated user.""" + terms_config = get_terms_of_use_config(settings) + if not terms_config["enabled"]: + return None + + if terms_config["frequency"] != "every_session" and _user_acceptance_matches(terms_config, user_id): + return None + + acceptance_record = _build_acceptance_record(terms_config) + if has_request_context(): + session[TERMS_OF_USE_SESSION_KEY] = acceptance_record + session.pop(TERMS_OF_USE_PRE_AUTH_SESSION_KEY, None) + session.modified = True + + if terms_config["frequency"] in {"daily", "once"}: + if not update_user_settings( + user_id, + {TERMS_OF_USE_USER_SETTINGS_KEY: acceptance_record}, + ): + raise RuntimeError("Terms of Use acceptance could not be saved.") + + log_terms_of_use_accepted( + user_id=user_id, + terms_hash=acceptance_record["hash"], + frequency=acceptance_record["frequency"], + source=source, + accepted_date=acceptance_record["accepted_date"], + auth_state="authenticated", + ) + return acceptance_record + + +def apply_pending_pre_auth_terms_of_use(user_id, settings, source="pre_auth"): + """Persist a matching pre-auth acceptance after authentication succeeds.""" + if not has_request_context(): + return None + + terms_config = get_terms_of_use_config(settings) + pending_acceptance = session.get(TERMS_OF_USE_PRE_AUTH_SESSION_KEY) + if not terms_config["enabled"] or not isinstance(pending_acceptance, dict): + return None + + if ( + pending_acceptance.get("hash") != terms_config["hash"] + or pending_acceptance.get("frequency") != terms_config["frequency"] + ): + session.pop(TERMS_OF_USE_PRE_AUTH_SESSION_KEY, None) + session.pop(TERMS_OF_USE_SESSION_KEY, None) + session.modified = True + return None + + if terms_config["frequency"] != "every_session" and _user_acceptance_matches(terms_config, user_id): + session.pop(TERMS_OF_USE_PRE_AUTH_SESSION_KEY, None) + session[TERMS_OF_USE_SESSION_KEY] = pending_acceptance + session.modified = True + return None + + if terms_config["frequency"] in {"daily", "once"}: + if not update_user_settings( + user_id, + {TERMS_OF_USE_USER_SETTINGS_KEY: pending_acceptance}, + ): + raise RuntimeError("Pre-auth terms of use acceptance could not be saved.") + + session[TERMS_OF_USE_SESSION_KEY] = pending_acceptance + session.pop(TERMS_OF_USE_PRE_AUTH_SESSION_KEY, None) + session.modified = True + log_terms_of_use_accepted( + user_id=user_id, + terms_hash=pending_acceptance["hash"], + frequency=pending_acceptance["frequency"], + source=source, + accepted_date=pending_acceptance["accepted_date"], + auth_state="authenticated", + ) + return pending_acceptance + + +def record_terms_of_use_decline(user_id, settings, source="post_auth"): + """Audit an authenticated decline of the current Terms of Use.""" + terms_config = get_terms_of_use_config(settings) + if not terms_config["enabled"] or not user_id: + return None + log_terms_of_use_declined( + user_id=user_id, + terms_hash=terms_config["hash"], + frequency=terms_config["frequency"], + source=source, + redirect_url=terms_config["decline_redirect_url"], + auth_state="authenticated", + ) + return terms_config + + +def log_pre_auth_terms_of_use_issue(message, extra=None): + """Log non-fatal pre-auth Terms of Use issues without exposing terms content.""" + log_event( + f"[TermsOfUse] {message}", + extra=extra or {}, + level=logging.WARNING, + ) diff --git a/application/single_app/route_frontend_admin_settings.py b/application/single_app/route_frontend_admin_settings.py index 751b0c1f..89795152 100644 --- a/application/single_app/route_frontend_admin_settings.py +++ b/application/single_app/route_frontend_admin_settings.py @@ -26,6 +26,15 @@ from functions_notifications import broadcast_system_notification from functions_logging import * from functions_document_actions import normalize_document_action_capabilities +from functions_terms_of_use import ( + TERMS_OF_USE_DEFAULT_REDIRECT, + TERMS_OF_USE_MAX_BUTTON_TEXT_LENGTH, + TERMS_OF_USE_MAX_MESSAGE_LENGTH, + TERMS_OF_USE_MAX_TITLE_LENGTH, + normalize_terms_of_use_frequency, + normalize_terms_of_use_redirect_url, + normalize_terms_of_use_text, +) from swagger_wrapper import swagger_route, get_auth_security from datetime import datetime, timedelta, timezone from admin_settings_int_utils import safe_int_with_source @@ -424,7 +433,23 @@ def admin_settings(): settings['classification_banner_color'] = '#ffc107' # Bootstrap warning color if 'classification_banner_text_color' not in settings: settings['classification_banner_text_color'] = '#ffffff' # White text by default - + + # --- Add defaults for terms of use --- + if 'enable_terms_of_use' not in settings: + settings['enable_terms_of_use'] = False + if 'terms_of_use_title' not in settings: + settings['terms_of_use_title'] = 'Terms of Use' + if 'terms_of_use_message' not in settings: + settings['terms_of_use_message'] = '' + settings['terms_of_use_frequency'] = normalize_terms_of_use_frequency( + settings.get('terms_of_use_frequency') + ) + if 'terms_of_use_decline_redirect_url' not in settings: + settings['terms_of_use_decline_redirect_url'] = TERMS_OF_USE_DEFAULT_REDIRECT + if 'terms_of_use_accept_button_text' not in settings: + settings['terms_of_use_accept_button_text'] = 'Accept and continue' + if 'terms_of_use_decline_button_text' not in settings: + settings['terms_of_use_decline_button_text'] = 'Cancel' # --- Add defaults for user agreement --- if 'enable_user_agreement' not in settings: settings['enable_user_agreement'] = False @@ -1605,6 +1630,47 @@ def parse_admin_int(raw_value, fallback_value, field_name="unknown", hard_defaul if word_count > 200: flash('User Agreement text exceeds 200 word limit. Please shorten the text.', 'warning') + # --- Terms of Use Settings --- + enable_terms_of_use = form_data.get('enable_terms_of_use') == 'on' + terms_of_use_title = normalize_terms_of_use_text( + form_data.get('terms_of_use_title'), + fallback='Terms of Use', + max_length=TERMS_OF_USE_MAX_TITLE_LENGTH, + ) + terms_of_use_message = normalize_terms_of_use_text( + form_data.get('terms_of_use_message'), + max_length=TERMS_OF_USE_MAX_MESSAGE_LENGTH, + ) + terms_of_use_frequency = normalize_terms_of_use_frequency( + form_data.get('terms_of_use_frequency') + ) + terms_of_use_accept_button_text = normalize_terms_of_use_text( + form_data.get('terms_of_use_accept_button_text'), + fallback='Accept and continue', + max_length=TERMS_OF_USE_MAX_BUTTON_TEXT_LENGTH, + ) + terms_of_use_decline_button_text = normalize_terms_of_use_text( + form_data.get('terms_of_use_decline_button_text'), + fallback='Cancel', + max_length=TERMS_OF_USE_MAX_BUTTON_TEXT_LENGTH, + ) + raw_terms_of_use_decline_redirect_url = form_data.get( + 'terms_of_use_decline_redirect_url', + TERMS_OF_USE_DEFAULT_REDIRECT, + ) + terms_of_use_decline_redirect_url = normalize_terms_of_use_redirect_url( + raw_terms_of_use_decline_redirect_url + ) + if ( + raw_terms_of_use_decline_redirect_url.strip() + and terms_of_use_decline_redirect_url != raw_terms_of_use_decline_redirect_url.strip() + ): + flash('Terms of Use cancel redirect was invalid and has been reset to the default.', 'warning') + + if enable_terms_of_use and not terms_of_use_message: + flash('Terms of Use message is required when the feature is enabled.', 'danger') + return redirect(url_for('frontend_admin_settings.admin_settings')) + # --- Authentication & Redirect Settings --- enable_front_door = form_data.get('enable_front_door') == 'on' front_door_url = form_data.get('front_door_url', '').strip() @@ -1976,6 +2042,15 @@ def is_valid_url(url): 'user_agreement_apply_to': user_agreement_apply_to, 'enable_user_agreement_daily': enable_user_agreement_daily, + # Terms of Use + 'enable_terms_of_use': enable_terms_of_use, + 'terms_of_use_title': terms_of_use_title, + 'terms_of_use_message': terms_of_use_message, + 'terms_of_use_frequency': terms_of_use_frequency, + 'terms_of_use_decline_redirect_url': terms_of_use_decline_redirect_url, + 'terms_of_use_accept_button_text': terms_of_use_accept_button_text, + 'terms_of_use_decline_button_text': terms_of_use_decline_button_text, + # Multimedia & Metadata 'enable_video_file_support': enable_video_file_support, 'enable_audio_file_support': enable_audio_file_support, diff --git a/application/single_app/route_frontend_authentication.py b/application/single_app/route_frontend_authentication.py index bce84d02..b3cd363b 100644 --- a/application/single_app/route_frontend_authentication.py +++ b/application/single_app/route_frontend_authentication.py @@ -7,6 +7,11 @@ from config import * from functions_activity_logging import log_user_login, record_user_login_session_activity +from functions_terms_of_use import ( + apply_pending_pre_auth_terms_of_use, + get_terms_of_use_config, + has_terms_of_use_acceptance, +) from functions_appinsights import log_event from functions_authentication import _build_msal_app, _load_cache, _save_cache, clear_requested_oauth_scopes, create_ci_bearer_session, get_graph_authority, get_graph_endpoint, get_requested_oauth_scopes from functions_debug import debug_print @@ -165,7 +170,11 @@ def login(): # Get settings from database, with environment variable fallback settings = get_settings() or {} - + + terms_config = get_terms_of_use_config(settings) + if terms_config["enabled"] and not has_terms_of_use_acceptance(settings): + return redirect(url_for('frontend_terms_of_use.terms_of_use')) + # Only use Front Door redirect URL if Front Door is enabled if settings.get('enable_front_door', False): front_door_url = settings.get('front_door_url') @@ -264,6 +273,22 @@ def authorized(): record_user_login_session_activity(session) except Exception as e: debug_print(f"Could not log login activity: {e}") + + try: + user_id = session['user'].get('oid') or session['user'].get('sub') + if user_id: + apply_pending_pre_auth_terms_of_use( + user_id=user_id, + settings=settings, + source="azure_ad_pre_auth", + ) + except Exception as e: + log_event( + "[TermsOfUse] Failed to apply pending Azure AD pre-auth acceptance.", + extra={'error': str(e)}, + level=logging.ERROR, + exceptionTraceback=True, + ) # Redirect to the originally intended page or home # You might want to store the original destination in the session during /login @@ -420,6 +445,20 @@ def teams_token_exchange(): except Exception as e: debug_print(f"[TeamsSSO] Could not log Teams login activity: {e}") + try: + apply_pending_pre_auth_terms_of_use( + user_id=session_user.get('oid'), + settings=get_settings() or {}, + source="teams_sso_pre_auth", + ) + except Exception as e: + log_event( + "[TermsOfUse] Failed to apply pending Teams pre-auth acceptance.", + extra={'error': str(e)}, + level=logging.ERROR, + exceptionTraceback=True, + ) + return jsonify({ "success": True, "user": { diff --git a/application/single_app/route_frontend_terms_of_use.py b/application/single_app/route_frontend_terms_of_use.py new file mode 100644 index 00000000..b3b57386 --- /dev/null +++ b/application/single_app/route_frontend_terms_of_use.py @@ -0,0 +1,115 @@ +# route_frontend_terms_of_use.py + +import logging + +from flask import redirect, render_template, request, session, url_for + +from functions_appinsights import log_event +from functions_terms_of_use import ( + TERMS_OF_USE_RETURN_PATH_SESSION_KEY, + get_terms_of_use_config, + has_terms_of_use_acceptance, + mark_pre_auth_terms_of_use_acceptance, + normalize_terms_of_use_return_path, + record_terms_of_use_acceptance, + record_terms_of_use_decline, +) +from functions_authentication import get_current_user_id +from functions_settings import get_settings, sanitize_settings_for_user +from swagger_wrapper import get_auth_security, swagger_route + + +def _store_terms_of_use_return_path(raw_return_path): + """Store a local-only post-acceptance return path in the server-side session.""" + return_path = normalize_terms_of_use_return_path( + raw_return_path, + fallback=url_for('public_app.index'), + ) + session[TERMS_OF_USE_RETURN_PATH_SESSION_KEY] = return_path + session.modified = True + return return_path + + +def _pop_terms_of_use_return_path(): + """Resolve and clear the local-only return path saved during the interstitial GET.""" + return_path = normalize_terms_of_use_return_path( + session.pop(TERMS_OF_USE_RETURN_PATH_SESSION_KEY, None), + fallback=url_for('public_app.index'), + ) + session.modified = True + return return_path + + +def register_route_frontend_terms_of_use(bp): + @bp.route('/terms-of-use', methods=['GET']) + @swagger_route(security=get_auth_security()) + def terms_of_use(): + settings = get_settings() or {} + terms_config = get_terms_of_use_config(settings) + _store_terms_of_use_return_path(request.args.get('next')) + + if not terms_config["enabled"]: + return redirect(url_for('public_app.index')) + + user_id = get_current_user_id() + if user_id and has_terms_of_use_acceptance(settings, user_id=user_id): + return redirect(url_for('public_app.index')) + + public_settings = sanitize_settings_for_user(settings) + return render_template( + 'terms_of_use.html', + app_settings=public_settings, + terms=terms_config, + is_authenticated=bool(user_id), + ) + + @bp.route('/terms-of-use/accept', methods=['POST']) + @swagger_route(security=get_auth_security()) + def accept_terms_of_use(): + settings = get_settings() or {} + terms_config = get_terms_of_use_config(settings) + return_url = _pop_terms_of_use_return_path() + + if not terms_config["enabled"]: + return redirect(url_for('public_app.index')) + + user_id = get_current_user_id() + if user_id: + record_terms_of_use_acceptance( + user_id=user_id, + settings=settings, + source="post_auth", + ) + return redirect(return_url) + + mark_pre_auth_terms_of_use_acceptance(settings) + return redirect(url_for('frontend_authentication.login')) + + @bp.route('/terms-of-use/decline', methods=['POST']) + @swagger_route(security=get_auth_security()) + def decline_terms_of_use(): + settings = get_settings() or {} + terms_config = get_terms_of_use_config(settings) + user_id = get_current_user_id() + + if user_id: + try: + record_terms_of_use_decline( + user_id=user_id, + settings=settings, + source="post_auth", + ) + except Exception as decline_log_error: + log_event( + "[TermsOfUse] Decline audit logging failed.", + extra={ + "user_id": user_id, + "error": str(decline_log_error), + }, + level=logging.ERROR, + exceptionTraceback=True, + ) + + redirect_url = terms_config["decline_redirect_url"] + session.clear() + return redirect(redirect_url) diff --git a/application/single_app/templates/admin_settings.html b/application/single_app/templates/admin_settings.html index 3e24e4e6..e9b75821 100644 --- a/application/single_app/templates/admin_settings.html +++ b/application/single_app/templates/admin_settings.html @@ -2194,6 +2194,107 @@
+
+
+ Terms of Use +
+

+ Require users to accept the Terms of Use before using SimpleChat. Standard Microsoft sign-in users see it before authentication; Teams SSO and other passive sign-in flows are gated immediately after the app session is created. +

+ +
+ + + +
+ +
+
+ + +
+
+ + + Changing the title, message, or frequency creates a new terms version that users must accept again. +
+
+ +
+ + + Plain text is shown to users with line breaks preserved. +
+ +
+ + + Use a local path such as / or an admin-approved HTTP(S) URL. Signed-in users are locally logged out before this redirect. +
+ +
+
+ + +
+
+ + +
+
+ + +
+
Support @@ -6595,9 +6696,9 @@
- {% endif %} + {% endif %}
diff --git a/application/single_app/templates/terms_of_use.html b/application/single_app/templates/terms_of_use.html new file mode 100644 index 00000000..878d1ffa --- /dev/null +++ b/application/single_app/templates/terms_of_use.html @@ -0,0 +1,118 @@ + + + + + + + {{ terms.title }} - {{ app_settings.app_title or 'SimpleChat' }} + + + + + +
+ + +

{{ terms.title }}

+
{{ terms.message }}
+ +
+
+ +
+
+ +
+
+
+ + diff --git a/docs/explanation/features/TERMS_OF_USE.md b/docs/explanation/features/TERMS_OF_USE.md new file mode 100644 index 00000000..3456b88d --- /dev/null +++ b/docs/explanation/features/TERMS_OF_USE.md @@ -0,0 +1,48 @@ +# Terms of Use + +## Overview + +Implemented in version: **0.250.055** +Current documentation version: **0.250.057** + +Terms of Use lets administrators require users to accept configurable terms, rules of behavior, or an entry notice before using SimpleChat. It is separate from the existing upload-focused User Agreement. + +## Technical Specifications + +* **Admin settings**: General tab settings control enablement, title, message, recurrence, cancel redirect, and button labels. +* **Recurrence options**: + * Every session: stored in the Flask session. + * Once per day: stored in user settings with the accepted UTC date. + * Just once: stored in user settings for the current terms version. +* **Terms versioning**: A hash of the title, message, and frequency invalidates older acceptances when admins change the Terms of Use. +* **Authentication integration**: + * Standard Microsoft sign-in users see the Terms of Use before being sent to Entra ID. + * SSO/passive sign-in users are gated immediately after the SimpleChat session is created. +* **Redirect safety**: User-controlled return paths are local-only and stored server-side in the session. The decline destination is admin-configured and may be local or an HTTPS URL. +* **Server-side enforcement**: Authenticated browser requests are redirected to the Terms of Use page until accepted. Authenticated API requests receive a `403` response with `terms_of_use_required`. +* **Audit logging**: Accepted and declined events are written to activity logs when a user identity is known. + +## Usage Instructions + +1. Open **Admin Settings**. +2. Select the **General** tab. +3. Enable **Require terms of use**. +4. Enter the Terms of Use title and message. +5. Choose the recurrence: + * **At the start of every session** + * **Once per day** + * **Just once per terms version** +6. Configure the cancel redirect URL. Use a local path such as `/` or an admin-approved HTTP(S) URL. +7. Save settings. + +Users who decline are logged out locally and redirected to the configured cancel destination. + +## Testing and Validation + +* Functional coverage: `functional_tests/test_terms_of_use.py` +* Route policy coverage: `functional_tests/route_tests/` +* UI template coverage: `ui_tests/test_terms_of_use_ui.py` + +## Known Limitations + +Before standard authentication, SimpleChat cannot know which user is signing in. Daily and once-per-version persistence is therefore applied after authentication, while the pre-auth prompt is tracked in the anonymous Flask session. diff --git a/docs/explanation/release_notes.md b/docs/explanation/release_notes.md index 56351e4f..b6ffca3b 100644 --- a/docs/explanation/release_notes.md +++ b/docs/explanation/release_notes.md @@ -2,6 +2,25 @@ For feature-focused and fix-focused drill-downs by version, see [Features by Version](/explanation/features/) and [Fixes by Version](/explanation/fixes/). +### **(v0.250.057)** + +#### Bug Fixes + +* **Terms of Use Redirect Hardening** + * Replaced wildcard config imports in the Terms of Use route with explicit Flask imports. + * Moved post-acceptance return paths from hidden form values to server-side session storage, keeping user-controlled return targets local-only. + * Restricted admin-configured external decline redirects to HTTPS URLs without embedded credentials while preserving local-path redirects. + * (Ref: microsoft/simplechat#504, `route_frontend_terms_of_use.py`, `functions_terms_of_use.py`, `terms_of_use.html`) + +### **(v0.250.056)** + +#### New Features + +* **Optional Terms of Use Gate** + * Added an admin-configurable Terms of Use prompt that can require users to accept rules of behavior, terms, or an entry notice before using SimpleChat. + * Supports every-session, once-per-day, and once-per-version recurrence modes, with server-side browser/API enforcement and activity logging for accept/decline events. + * (Ref: microsoft/simplechat#504, `TERMS_OF_USE.md`, `functions_terms_of_use.py`, `route_frontend_terms_of_use.py`, `terms_of_use.html`) + ### **(v0.250.052)** #### Bug Fixes diff --git a/functional_tests/route_tests/test_route_blueprint_policy_inventory.py b/functional_tests/route_tests/test_route_blueprint_policy_inventory.py index b236df2b..c07eafea 100644 --- a/functional_tests/route_tests/test_route_blueprint_policy_inventory.py +++ b/functional_tests/route_tests/test_route_blueprint_policy_inventory.py @@ -2,7 +2,7 @@ # test_route_blueprint_policy_inventory.py """ Functional test for route blueprint policy inventory. -Version: 0.250.006 +Version: 0.250.055 Implemented in: 0.242.069 This test ensures every SimpleChat route is assigned to a Blueprint-based @@ -71,6 +71,7 @@ "external_no_auth_health": (), "frontend_admin_settings": ("login_required", "admin_required"), "frontend_agents": ("login_required", "user_required"), + "frontend_terms_of_use": (), "frontend_authentication": (), "frontend_chats": ("login_required", "user_required"), "frontend_control_center": ("login_required",), @@ -94,6 +95,9 @@ "/getATokenApi", "/logout", "/logout/local", + "/terms-of-use", + "/terms-of-use/accept", + "/terms-of-use/decline", "/ci-auth/session", "/robots933456.txt", "/favicon.ico", diff --git a/functional_tests/route_tests/test_route_policy_test_coverage.py b/functional_tests/route_tests/test_route_policy_test_coverage.py index f7983539..f765df58 100644 --- a/functional_tests/route_tests/test_route_policy_test_coverage.py +++ b/functional_tests/route_tests/test_route_policy_test_coverage.py @@ -2,7 +2,7 @@ # test_route_policy_test_coverage.py """ Functional test for route policy test coverage completeness. -Version: 0.250.003 +Version: 0.250.055 Implemented in: 0.242.069 This test ensures the route inventory and unauthenticated access policy tests diff --git a/functional_tests/route_tests/test_route_unauthenticated_policy_contract.py b/functional_tests/route_tests/test_route_unauthenticated_policy_contract.py index d9de287c..1a007aa0 100644 --- a/functional_tests/route_tests/test_route_unauthenticated_policy_contract.py +++ b/functional_tests/route_tests/test_route_unauthenticated_policy_contract.py @@ -2,7 +2,7 @@ # test_route_unauthenticated_policy_contract.py """ Functional test for route unauthenticated access policy contract. -Version: 0.250.003 +Version: 0.250.055 Implemented in: 0.242.069 This test ensures every SimpleChat route has an explicit expected unauthenticated @@ -25,6 +25,9 @@ "/getATokenApi", "/logout", "/logout/local", + "/terms-of-use", + "/terms-of-use/accept", + "/terms-of-use/decline", "/ci-auth/session", "/robots933456.txt", "/favicon.ico", diff --git a/functional_tests/test_terms_of_use.py b/functional_tests/test_terms_of_use.py new file mode 100644 index 00000000..1c074364 --- /dev/null +++ b/functional_tests/test_terms_of_use.py @@ -0,0 +1,216 @@ +#!/usr/bin/env python3 +# test_terms_of_use.py +""" +Functional test for Terms of Use recurrence and persistence helpers. +Version: 0.250.057 +Implemented in: 0.250.055 +Redirect hardening updated in: 0.250.057 + +This test ensures Terms of Use hashes, redirect validation, +pre-auth session acceptance, and user-settings persistence behave consistently. +""" + +import os +import sys +import types +import importlib.util +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[1] +APP_DIR = REPO_ROOT / "application" / "single_app" + + +class FakeSession(dict): + """Minimal Flask session stand-in for helper tests.""" + + modified = False + + +fake_session = FakeSession() +_MISSING_MODULE = object() + + +def _install_helper_test_stubs(): + fake_flask = types.ModuleType("flask") + fake_flask.session = fake_session + fake_flask.has_request_context = lambda: True + stubs = {"flask": fake_flask} + + fake_activity = types.ModuleType("functions_activity_logging") + fake_activity.log_terms_of_use_accepted = lambda **payload: None + fake_activity.log_terms_of_use_declined = lambda **payload: None + stubs["functions_activity_logging"] = fake_activity + + fake_appinsights = types.ModuleType("functions_appinsights") + fake_appinsights.log_event = lambda *args, **kwargs: None + stubs["functions_appinsights"] = fake_appinsights + + fake_settings = types.ModuleType("functions_settings") + fake_settings.get_user_settings = lambda user_id: {"id": user_id, "settings": {}} + fake_settings.update_user_settings = lambda user_id, payload: True + stubs["functions_settings"] = fake_settings + + originals = {} + for name, module in stubs.items(): + originals[name] = sys.modules.get(name, _MISSING_MODULE) + sys.modules[name] = module + return originals + + +def _restore_helper_test_stubs(originals): + for name, original_module in originals.items(): + if original_module is _MISSING_MODULE: + sys.modules.pop(name, None) + else: + sys.modules[name] = original_module + + +def _load_helper_module(): + originals = _install_helper_test_stubs() + module_path = APP_DIR / "functions_terms_of_use.py" + spec = importlib.util.spec_from_file_location("terms_of_use_under_test", module_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(module) + finally: + _restore_helper_test_stubs(originals) + return module + + +terms = _load_helper_module() + + +def _enabled_settings(frequency="once", message="Please accept these terms."): + return { + "enable_terms_of_use": True, + "terms_of_use_title": "Rules of Behavior", + "terms_of_use_message": message, + "terms_of_use_frequency": frequency, + "terms_of_use_decline_redirect_url": "/", + "terms_of_use_accept_button_text": "Accept", + "terms_of_use_decline_button_text": "Cancel", + } + + +def test_hash_and_redirect_normalization(): + """Validate hash invalidation and redirect safety.""" + first_hash = terms.compute_terms_of_use_hash("Title", "Message", "once") + second_hash = terms.compute_terms_of_use_hash("Title", "Changed", "once") + + assert first_hash != second_hash + assert terms.normalize_terms_of_use_frequency("once-per-day") == "daily" + assert terms.normalize_terms_of_use_redirect_url("/goodbye") == "/goodbye" + assert terms.normalize_terms_of_use_redirect_url("https://contoso.example/terms") == "https://contoso.example/terms" + assert terms.normalize_terms_of_use_redirect_url("http://contoso.example/terms") == "/" + assert terms.normalize_terms_of_use_redirect_url("https://user:pass@contoso.example/terms") == "/" + assert terms.normalize_terms_of_use_redirect_url("//evil.example") == "/" + assert terms.normalize_terms_of_use_redirect_url("javascript:alert(1)") == "/" + assert terms.normalize_terms_of_use_return_path("/chats?x=1") == "/chats?x=1" + assert terms.normalize_terms_of_use_return_path("https://evil.example") == "/" + + +def test_pre_auth_session_acceptance_unblocks_login_for_daily_mode(): + """Validate anonymous pre-auth acceptance prevents a daily-mode login loop.""" + settings = _enabled_settings(frequency="daily") + + fake_session.clear() + assert not terms.has_terms_of_use_acceptance(settings) + terms.mark_pre_auth_terms_of_use_acceptance(settings) + assert terms.has_terms_of_use_acceptance(settings) + assert terms.TERMS_OF_USE_PRE_AUTH_SESSION_KEY in fake_session + + +def test_once_acceptance_persists_to_user_settings_and_activity_log(): + """Validate once-per-version acceptance writes user settings and audit data.""" + settings = _enabled_settings(frequency="once") + updates = [] + audits = [] + + original_update_user_settings = terms.update_user_settings + original_get_user_settings = terms.get_user_settings + original_log_acceptance = terms.log_terms_of_use_accepted + + terms.update_user_settings = lambda user_id, payload: updates.append((user_id, payload)) or True + terms.get_user_settings = lambda user_id: {"id": user_id, "settings": {}} + terms.log_terms_of_use_accepted = lambda **payload: audits.append(payload) + + try: + fake_session.clear() + record = terms.record_terms_of_use_acceptance( + user_id="user-123", + settings=settings, + source="post_auth", + ) + assert record["frequency"] == "once" + assert fake_session[terms.TERMS_OF_USE_SESSION_KEY]["hash"] == record["hash"] + finally: + terms.update_user_settings = original_update_user_settings + terms.get_user_settings = original_get_user_settings + terms.log_terms_of_use_accepted = original_log_acceptance + + assert updates[0][0] == "user-123" + stored_record = updates[0][1][terms.TERMS_OF_USE_USER_SETTINGS_KEY] + assert stored_record["hash"] == record["hash"] + assert audits[0]["user_id"] == "user-123" + assert audits[0]["frequency"] == "once" + + +def test_daily_user_settings_acceptance_requires_today(): + """Validate daily recurrence expires on the next UTC date.""" + settings = _enabled_settings(frequency="daily") + config = terms.get_terms_of_use_config(settings) + today_record = { + "hash": config["hash"], + "frequency": "daily", + "accepted_date": terms._utc_now().strftime("%Y-%m-%d"), + } + stale_record = { + "hash": config["hash"], + "frequency": "daily", + "accepted_date": "2000-01-01", + } + + original_get_user_settings = terms.get_user_settings + try: + terms.get_user_settings = lambda user_id: { + "id": user_id, + "settings": {terms.TERMS_OF_USE_USER_SETTINGS_KEY: today_record}, + } + assert terms.has_terms_of_use_acceptance(settings, user_id="user-123") + + terms.get_user_settings = lambda user_id: { + "id": user_id, + "settings": {terms.TERMS_OF_USE_USER_SETTINGS_KEY: stale_record}, + } + assert not terms.has_terms_of_use_acceptance(settings, user_id="user-123") + finally: + terms.get_user_settings = original_get_user_settings + + +if __name__ == "__main__": + os.environ.setdefault("DISABLE_FLASK_INSTRUMENTATION", "1") + tests = [ + test_hash_and_redirect_normalization, + test_pre_auth_session_acceptance_unblocks_login_for_daily_mode, + test_once_acceptance_persists_to_user_settings_and_activity_log, + test_daily_user_settings_acceptance_requires_today, + ] + results = [] + for test in tests: + print(f"\nRunning {test.__name__}...") + try: + test() + print("PASS") + results.append(True) + except Exception as exc: + print(f"FAIL: {exc}") + import traceback + + traceback.print_exc() + results.append(False) + + passed = sum(results) + print(f"\nResults: {passed}/{len(results)} tests passed") + sys.exit(0 if all(results) else 1) diff --git a/ui_tests/test_admin_cosmos_throughput_settings_ui.py b/ui_tests/test_admin_cosmos_throughput_settings_ui.py index b23e2e19..3a1b8881 100644 --- a/ui_tests/test_admin_cosmos_throughput_settings_ui.py +++ b/ui_tests/test_admin_cosmos_throughput_settings_ui.py @@ -2,7 +2,7 @@ """ UI test for Admin Settings Cosmos throughput controls. -Version: 0.250.045 +Version: 0.250.055 Implemented in: 0.241.147 This test ensures the Scale tab exposes Cosmos throughput monitoring and @@ -22,8 +22,10 @@ Version 0.241.184 adds neutral informational copy for normal container-targeted throughput mode. Version 0.241.199 adds SimpleChat's 10,000 RU/s scale support ceiling, monitor-only indicators, and container policy modal filtering coverage. Version 0.250.045 adds a stable left-nav anchor for the Cosmos metrics table. +Version 0.250.054 adds regression coverage that Scale-only sections remain inside the Scale tab when DAI debug UI is disabled. """ +from html.parser import HTMLParser import re from pathlib import Path @@ -41,6 +43,68 @@ ADMIN_JS = REPO_ROOT / "application" / "single_app" / "static" / "js" / "admin" / "admin_settings.js" +class _TabPaneAncestorParser(HTMLParser): + """Track tab-pane ancestors for selected elements after template pruning.""" + + _void_tags = { + "area", + "base", + "br", + "col", + "embed", + "hr", + "img", + "input", + "link", + "meta", + "param", + "source", + "track", + "wbr", + } + + def __init__(self, target_ids): + super().__init__(convert_charrefs=False) + self.target_ids = set(target_ids) + self.stack = [] + self.pane_ancestors = {} + + def handle_starttag(self, tag, attrs): + attr_map = dict(attrs) + element_id = attr_map.get("id") + classes = (attr_map.get("class") or "").split() + + if element_id in self.target_ids: + pane_ancestors = [ + item["id"] + for item in self.stack + if item["tag"] == "div" and "tab-pane" in item["classes"] + ] + if tag == "div" and "tab-pane" in classes: + pane_ancestors.append(element_id) + self.pane_ancestors[element_id] = pane_ancestors + + if tag not in self._void_tags: + self.stack.append({ + "tag": tag, + "id": element_id, + "classes": classes, + }) + + def handle_endtag(self, tag): + for index in range(len(self.stack) - 1, -1, -1): + if self.stack[index]["tag"] == tag: + del self.stack[index:] + return + + +def _render_template_with_dai_debug_disabled(template): + modal_start = template.index('