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

1""" 

2Authentication decorators for protecting routes. 

3""" 

4 

5from functools import wraps 

6 

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

8from loguru import logger 

9 

10from ...database.encrypted_db import db_manager 

11 

12 

13def login_required(f): 

14 """ 

15 Decorator to require authentication for a route. 

16 Redirects to login page if not authenticated. 

17 """ 

18 

19 @wraps(f) 

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

21 if "username" not in session: 

22 logger.debug( 

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

24 ) 

25 # For API routes, return JSON error instead of redirect 

26 if request.path.startswith("/api/") or request.path.startswith( 

27 "/settings/api/" 

28 ): 

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

30 return redirect(url_for("auth.login", next=request.url)) 

31 

32 # Check if we have an active database connection 

33 username = session["username"] 

34 if not db_manager.connections.get(username): 

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

36 logger.debug( 

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

38 ) 

39 # For API routes, return JSON error instead of redirect 

40 if request.path.startswith("/api/") or request.path.startswith( 

41 "/settings/api/" 

42 ): 

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

44 session.clear() 

45 return redirect(url_for("auth.login", next=request.url)) 

46 

47 return f(*args, **kwargs) 

48 

49 return decorated_function 

50 

51 

52def current_user(): 

53 """ 

54 Get the current authenticated user's username. 

55 Returns None if not authenticated. 

56 """ 

57 return session.get("username") 

58 

59 

60def get_current_db_session(): 

61 """ 

62 Get the database session for the current user. 

63 Must be called within a login_required route. 

64 """ 

65 username = current_user() 

66 if username: 

67 return db_manager.get_session(username) 

68 return None 

69 

70 

71def inject_current_user(): 

72 """ 

73 Flask before_request handler to inject current user into g. 

74 """ 

75 g.current_user = current_user() 

76 if g.current_user: 

77 # Try to get the database session 

78 try: 

79 g.db_session = db_manager.get_session(g.current_user) 

80 if g.db_session is None: 

81 # Check if we have an active database connection for this user 

82 if not db_manager.connections.get(g.current_user): 82 ↛ exitline 82 didn't return from function 'inject_current_user' because the condition on line 82 was always true

83 # For authenticated users without a database connection, 

84 # we need to handle this differently based on the route type 

85 

86 # For API routes and auth routes, allow the request to continue 

87 # The individual route handlers will deal with the missing database 

88 if ( 88 ↛ 98line 88 didn't jump to line 98 because the condition on line 88 was always true

89 request.path.startswith("/api/") 

90 or request.path.startswith("/auth/") 

91 or request.path.startswith("/settings/api/") 

92 ): 

93 logger.debug( 

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

95 ) 

96 else: 

97 # For regular routes, this is a stale session that needs clearing 

98 logger.debug( 

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

100 ) 

101 session.clear() 

102 g.current_user = None 

103 g.db_session = None 

104 except Exception as e: 

105 logger.exception( 

106 f"Error getting session for user {g.current_user}: {e}" 

107 ) 

108 g.db_session = None 

109 else: 

110 g.db_session = None