Coverage for app/backend/src/couchers/rate_limits/check.py: 100%

1import logging 

2from collections.abc import Sequence 

3 

4from sqlalchemy import RowMapping, exists, select 

5from sqlalchemy.orm import Session 

6 

7from couchers.db import session_scope 

8from couchers.models import RateLimitAction, RateLimitViolation 

9from couchers.rate_limits.definitions import RATE_LIMIT_DEFINITIONS, RATE_LIMIT_INTERVAL 

10from couchers.tasks import send_rate_limit_violation_report_email 

11from couchers.utils import now 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16def _get_user_events_in_past_time_interval( 

17 session: Session, user_id: int 

18) -> dict[RateLimitAction, Sequence[RowMapping]]: 

19 """Get all relevant events for the user in the last rate limit interval for the mod email.""" 

20 return { 

21 action: RATE_LIMIT_DEFINITIONS[action].mod_email_information_query(session, user_id) 

22 for action in RateLimitAction 

23 } 

24 

25 

26def _save_rate_limit_violation( 

27 session: Session, user_id: int, action: RateLimitAction, is_hard_limit: bool 

28) -> RateLimitViolation: 

29 """Save a rate limit violation to the database and return it.""" 

30 violation = RateLimitViolation( 

31 user_id=user_id, 

32 action=action, 

33 is_hard_limit=is_hard_limit, 

34 ) 

35 session.add(violation) 

36 session.flush() 

37 return violation 

38 

39 

40def _user_has_violated_rate_limit_in_past_time_interval( 

41 session: Session, user_id: int, action: RateLimitAction, is_hard_limit: bool 

42) -> bool: 

43 """Check if a RateLimitViolation for the user for the given action exists in the last RATE_LIMIT_INTERVAL.""" 

44 return session.execute( 

45 select( 

46 exists().where( 

47 RateLimitViolation.user_id == user_id, 

48 RateLimitViolation.action == action, 

49 RateLimitViolation.created >= now() - RATE_LIMIT_INTERVAL, 

50 RateLimitViolation.is_hard_limit == is_hard_limit, 

51 ) 

52 ) 

53 ).scalar_one() 

54 

55 

56def process_rate_limits_and_check_abort(session: Session, user_id: int, action: RateLimitAction) -> bool: 

57 """ 

58 Check if the user has reached a rate limit. Notify the moderation team in a separate background job if so. 

59 

60 Returns True if the user has reached a hard rate limit. 

61 """ 

62 rate_limit_definition = RATE_LIMIT_DEFINITIONS[action] 

63 count_last_interval = rate_limit_definition.count_actions_query(session, user_id) 

64 for limit, is_hard_limit in [ 

65 (rate_limit_definition.hard_limit, True), 

66 (rate_limit_definition.warning_limit, False), 

67 ]: 

68 if count_last_interval >= limit: 

69 if not _user_has_violated_rate_limit_in_past_time_interval( 

70 session=session, user_id=user_id, action=action, is_hard_limit=is_hard_limit 

71 ): 

72 # the caller aborts on a hard limit, which rolls back its session, so record this separately 

73 with session_scope() as violation_session: 

74 rate_limit_violation = _save_rate_limit_violation( 

75 session=violation_session, user_id=user_id, action=action, is_hard_limit=is_hard_limit 

76 ) 

77 events = _get_user_events_in_past_time_interval(session=violation_session, user_id=user_id) 

78 send_rate_limit_violation_report_email( 

79 session=violation_session, 

80 rate_limit_violation=rate_limit_violation, 

81 threshold=limit, 

82 events=events, 

83 ) 

84 return is_hard_limit 

85 return False