Coverage for utilities/request.py: 47%

1import warnings 

2from contextlib import ExitStack, contextmanager 

3from urllib.parse import urlparse 

4 

5from django.conf import settings 

6from django.utils.datastructures import MultiValueDict 

7from django.utils.http import url_has_allowed_host_and_scheme 

8from django.utils.translation import gettext_lazy as _ 

9from netaddr import AddrFormatError, IPAddress 

10 

11from netbox.registry import registry 

12 

13from .constants import HTTP_REQUEST_META_SAFE_COPY, HTTP_REQUEST_META_SENSITIVE 

14 

15__all__ = ( 

16 'NetBoxFakeRequest', 

17 'apply_request_processors', 

18 'copy_safe_request', 

19 'get_client_ip', 

20 'get_safe_request_context', 

21 'safe_for_redirect', 

22) 

23 

24 

25# 

26# Fake request object 

27# 

28 

29class NetBoxFakeRequest: 

30 """ 

31 A fake request object which is explicitly defined at the module level so it is able to be pickled. It simply 

32 takes what is passed to it as kwargs on init and sets them as instance variables. 

33 """ 

34 def __init__(self, _dict): 

35 self.__dict__ = _dict 

36 

37 

38# 

39# Utility functions 

40# 

41 

42def copy_safe_request(request, include_files=True): 

43 """ 

44 Copy selected attributes from a request object into a new fake request object. This is needed in places where 

45 thread safe pickling of the useful request data is needed. 

46 

47 Args: 

48 request: The original request object 

49 include_files: Whether to include request.FILES. 

50 """ 

51 meta = {} 

52 for k, v in request.META.items(): 

53 if not isinstance(v, str): 

54 continue 

55 if k in HTTP_REQUEST_META_SAFE_COPY: 

56 meta[k] = v 

57 elif k.startswith('HTTP_') and k not in HTTP_REQUEST_META_SENSITIVE: 

58 meta[k] = v 

59 data = { 

60 'META': meta, 

61 'COOKIES': request.COOKIES, 

62 'POST': request.POST, 

63 'GET': request.GET, 

64 'user': request.user, 

65 'method': request.method, 

66 'path': request.path, 

67 'path_info': request.path_info, 

68 'id': getattr(request, 'id', None), # UUID assigned by middleware 

69 } 

70 if include_files: 

71 data['FILES'] = request.FILES 

72 else: 

73 data['FILES'] = MultiValueDict() 

74 

75 return NetBoxFakeRequest(data) 

76 

77 

78def get_safe_request_context(request): 

79 """ 

80 Return a sanitized subset of an HttpRequest suitable for exposure to user-authored templates 

81 (e.g. custom links). Excludes sensitive data such as cookies, headers, and session state. Returns a 

82 plain dict; Jinja2 resolves attribute access (e.g. request.path) against it via getitem fallback. 

83 """ 

84 if request is None: 

85 return None 

86 return { 

87 'id': str(request.id) if hasattr(request, 'id') else None, # UUID assigned by middleware 

88 'path': request.path, 

89 'path_info': request.path_info, 

90 'method': request.method, 

91 'GET': request.GET, 

92 'user': str(request.user), # Username only; not the User instance 

93 } 

94 

95 

96def get_client_ip(request, additional_headers=()): 

97 """ 

98 Return the client (source) IP address of the given request. Accepts an optional list of headers to inspect in 

99 addition to those configured under HTTP_CLIENT_IP_HEADERS. 

100 """ 

101 headers = ( 

102 *settings.HTTP_CLIENT_IP_HEADERS, 

103 *additional_headers, 

104 ) 

105 for header in headers: 105 ↛ 121line 105 didn't jump to line 121 because the loop on line 105 didn't complete

106 if header in request.META: 

107 ip = request.META[header].split(',')[0].strip() 

108 try: 

109 return IPAddress(ip) 

110 except AddrFormatError: 

111 # Parse the string with urlparse() to remove port number or any other cruft 

112 ip = urlparse(f'//{ip}').hostname 

113 

114 try: 

115 return IPAddress(ip) 

116 except AddrFormatError: 

117 # We did our best 

118 raise ValueError(_("Invalid IP address set for {header}: {ip}").format(header=header, ip=ip)) 

119 

120 # Could not determine the client IP address from request headers 

121 return None 

122 

123 

124def safe_for_redirect(url): 

125 """ 

126 Returns True if the given URL is safe to use as an HTTP redirect; otherwise returns False. 

127 """ 

128 return url_has_allowed_host_and_scheme(url, allowed_hosts=None) 

129 

130 

131@contextmanager 

132def apply_request_processors(request): 

133 """ 

134 A context manager with applies all registered request processors (such as event_tracking). 

135 """ 

136 with ExitStack() as stack: 

137 for request_processor in registry['request_processors']: 

138 try: 

139 stack.enter_context(request_processor(request)) 

140 except Exception as e: 

141 warnings.warn(f'Failed to initialize request processor {request_processor.__name__}: {e}') 

142 yield