Coverage for src/local_deep_research/web/auth/decorators.py: 100%

1""" 

2Authentication decorators for protecting routes. 

3""" 

4 

5from functools import wraps 

6from typing import Optional 

7 

8from flask import g, jsonify, redirect, request, session, url_for 

9from loguru import logger 

10from sqlalchemy.orm import Session 

11 

12from ...database.encrypted_db import db_manager 

13from ...security.url_validator import URLValidator 

14 

15 

16def _safe_redirect_to_login(): 

17 """ 

18 Redirect to login with validated next parameter. 

19 

20 Uses request.url as next parameter only if it passes 

21 security validation to prevent open redirect vulnerabilities. 

22 

23 Returns: 

24 Flask redirect response 

25 """ 

26 next_url = request.url 

27 # Validate that next URL is safe before using it 

28 if URLValidator.is_safe_redirect_url(next_url, request.host_url): 

29 return redirect(url_for("auth.login", next=next_url)) 

30 # Fall back to login without next parameter if validation fails 

31 return redirect(url_for("auth.login")) 

32 

33 

34def _is_api_path(path: str) -> bool: 

35 """Detect API request paths that should receive JSON, not HTML redirects. 

36 

37 Matches `/api/` anywhere in the path (so nested API blueprints like 

38 `/news/api/...` and `/library/api/...` work, not just top-level 

39 `/api/...`), and also paths that end in `/api` with no further 

40 segments (e.g. `/settings/api`, `/history/api` are JSON endpoints). 

41 

42 The `api` segment must be slash-bounded — non-API paths like 

43 `/apidocs` or `/openapi.json` are not matched. 

44 """ 

45 return "/api/" in path or path.endswith("/api") 

46 

47 

48def login_required(f): 

49 """ 

50 Decorator to require authentication for a route. 

51 Redirects to login page if not authenticated. 

52 """ 

53 

54 @wraps(f) 

55 def decorated_function(*args, **kwargs): 

56 if "username" not in session: 

57 logger.debug( 

58 f"Unauthenticated access attempt to {request.endpoint}" 

59 ) 

60 if _is_api_path(request.path): 

61 return jsonify({"error": "Authentication required"}), 401 

62 return _safe_redirect_to_login() 

63 

64 # Check if we have an active database connection 

65 username = session["username"] 

66 if not db_manager.is_user_connected(username): 

67 # Use debug level to reduce log noise for persistent sessions 

68 logger.debug( 

69 f"No database connection for authenticated user {username}" 

70 ) 

71 if _is_api_path(request.path): 

72 return jsonify({"error": "Database connection required"}), 401 

73 session.clear() 

74 return _safe_redirect_to_login() 

75 

76 return f(*args, **kwargs) 

77 

78 return decorated_function 

79 

80 

81def current_user(): 

82 """ 

83 Get the current authenticated user's username. 

84 Returns None if not authenticated. 

85 """ 

86 return session.get("username") 

87 

88 

89def get_current_db_session() -> Optional[Session]: 

90 """ 

91 Get the database session for the current user. 

92 Must be called within a login_required route. 

93 """ 

94 username = current_user() 

95 if username: 

96 return db_manager.get_session(username) 

97 return None 

98 

99 

100def inject_current_user(): 

101 """ 

102 Flask before_request handler to inject current user into g. 

103 """ 

104 g.current_user = current_user() 

105 if g.current_user: 

106 # Check connectivity 

107 if not db_manager.is_user_connected(g.current_user): 

108 # For API/auth routes, allow the request to continue 

109 if _is_api_path(request.path) or request.path.startswith("/auth/"): 

110 logger.debug( 

111 f"No database for user {g.current_user} on API/auth route" 

112 ) 

113 else: 

114 logger.debug( 

115 f"Clearing stale session for user {g.current_user}" 

116 ) 

117 session.clear() 

118 g.current_user = None 

119 g.db_session = None 

120 else: 

121 # Session will be created lazily by get_g_db_session() on first 

122 # access. This avoids checking out a pool connection for requests 

123 # that never touch the database (status polls, health checks, etc.). 

124 g.db_session = None 

125 else: 

126 g.db_session = None