Coverage for product_risk_suite/scraper/services.py: 85%

1import json 

2import logging 

3 

4import requests 

5from bs4 import BeautifulSoup 

6from django.contrib.auth.models import User 

7from django.utils import timezone 

8from requests.auth import HTTPBasicAuth 

9 

10from .models import AssigneeMapping, RemoteCredential, Scraper 

11from .pathutils import resolve_html_value, resolve_json_path 

12 

13logger = logging.getLogger("scraper") 

14 

15REQUEST_TIMEOUT_SECONDS = 10 

16 

17 

18def run_scraper_for_evidence(evidence): 

19 """Fetch and apply assignee/status for a single Evidence from its assigned 

20 Scraper. Returns True on success, False on failure, None when no scraper is 

21 assigned (no-op).""" 

22 scraper = evidence.scraper 

23 if scraper is None: 

24 return None 

25 

26 url = (evidence.scrape_url or evidence.evidence_link or "").strip() 

27 credential = scraper.credential 

28 if not url or (credential is not None and not url.startswith(credential.url)): 

29 logger.warning("evidence %s: no usable/matching scrape URL for scraper %s", evidence.pk, scraper.name) 

30 return _mark_failed(evidence) 

31 

32 try: 

33 response = _fetch(scraper, url) 

34 response.raise_for_status() 

35 except requests.RequestException as exc: 

36 logger.warning("evidence %s: request to %s failed: %s", evidence.pk, url, exc) 

37 return _mark_failed(evidence) 

38 

39 try: 

40 raw_assignee, raw_status = _extract(scraper, response) 

41 except (ValueError, KeyError, IndexError, TypeError, AttributeError) as exc: 

42 logger.warning("evidence %s: could not extract fields from response: %s", evidence.pk, exc) 

43 return _mark_failed(evidence) 

44 

45 status = _resolve_status(scraper, raw_status) 

46 if status is None: 

47 logger.warning("evidence %s: unmapped status value %r for scraper %s", evidence.pk, raw_status, scraper.name) 

48 return _mark_failed(evidence) 

49 

50 evidence.status = status 

51 

52 if raw_assignee: 

53 responsible = _resolve_responsible(url, raw_assignee) 

54 evidence.responsible = responsible 

55 if responsible is None: 

56 logger.warning("evidence %s: no user matches assignee %r, unassigning", evidence.pk, raw_assignee) 

57 

58 evidence.last_scrape_failed = False 

59 evidence.last_successful_scrape_at = timezone.now() 

60 evidence.save(update_fields=["status", "responsible", "last_scrape_failed", "last_successful_scrape_at"]) 

61 return True 

62 

63 

64def test_scraper(scraper, url): 

65 """Dry-run a scraper against a URL without touching any Evidence — used by the 

66 'Test scraper' admin preview. Returns a dict describing input, raw response and 

67 what would have been extracted/resolved; never raises.""" 

68 result = { 

69 "url": url, 

70 "credential": scraper.credential.name if scraper.credential else None, 

71 "response_format": scraper.get_response_format_display(), 

72 "error": None, 

73 "raw_response": None, 

74 "raw_assignee": None, 

75 "raw_status": None, 

76 "resolved_responsible": None, 

77 "resolved_status": None, 

78 } 

79 

80 if not url: 

81 result["error"] = "No URL to scrape ('Scrape url' and 'Evidence link' are both empty)." 

82 return result 

83 

84 try: 

85 response = _fetch(scraper, url) 

86 response.raise_for_status() 

87 except requests.RequestException as exc: 

88 result["error"] = f"Request failed: {exc}" 

89 return result 

90 

91 if scraper.response_format == Scraper.JSON: 

92 try: 

93 result["raw_response"] = json.dumps(response.json(), indent=2, ensure_ascii=False) 

94 except ValueError: 

95 result["raw_response"] = response.text 

96 else: 

97 result["raw_response"] = response.text 

98 

99 try: 

100 raw_assignee, raw_status = _extract(scraper, response) 

101 except (ValueError, KeyError, IndexError, TypeError, AttributeError) as exc: 

102 result["error"] = f"Could not extract fields from response: {exc}" 

103 return result 

104 

105 result["raw_assignee"] = raw_assignee 

106 result["raw_status"] = raw_status 

107 

108 status = _resolve_status(scraper, raw_status) 

109 result["resolved_status"] = str(status) if status is not None else None 

110 

111 responsible = _resolve_responsible(url, raw_assignee) 

112 result["resolved_responsible"] = str(responsible) if responsible is not None else None 

113 

114 return result 

115 

116 

117def _resolve_status(scraper, raw_status): 

118 if raw_status is None: 

119 return None 

120 mapping = scraper.status_mappings.filter(remote_value=raw_status).first() 

121 return mapping.status if mapping is not None else None 

122 

123 

124def _resolve_responsible(url, raw_assignee): 

125 if not raw_assignee: 

126 return None 

127 

128 mapped = _resolve_from_assignee_mapping(url, raw_assignee) 

129 if mapped is not None: 

130 return mapped 

131 

132 return ( 

133 User.objects.filter(email__iexact=raw_assignee).first() 

134 or User.objects.filter(username__iexact=raw_assignee).first() 

135 ) 

136 

137 

138def _resolve_from_assignee_mapping(url, raw_assignee): 

139 """Look up a per-host AssigneeMapping for the remote username. Only mappings 

140 whose base_url is a prefix of the scraped url are considered, so the same 

141 remote username on different hosts can map to different local users.""" 

142 for mapping in AssigneeMapping.objects.filter(remote_username__iexact=raw_assignee).select_related("user"): 

143 if url.startswith(mapping.base_url): 

144 return mapping.user 

145 return None 

146 

147 

148def _mark_failed(evidence): 

149 evidence.last_scrape_failed = True 

150 evidence.save(update_fields=["last_scrape_failed"]) 

151 return False 

152 

153 

154def _fetch(scraper, url): 

155 credential = scraper.credential 

156 headers = dict(scraper.extra_headers or {}) 

157 auth = None 

158 if credential is not None: 

159 if credential.auth_type == RemoteCredential.BASIC: 

160 auth = HTTPBasicAuth(credential.username, credential.password) 

161 else: 

162 headers.setdefault("Authorization", f"Bearer {credential.api_token}") 

163 return requests.get(url, auth=auth, headers=headers, timeout=REQUEST_TIMEOUT_SECONDS) 

164 

165 

166def _extract(scraper, response): 

167 if scraper.response_format == Scraper.JSON: 

168 data = response.json() 

169 return resolve_json_path(data, scraper.assignee_path), resolve_json_path(data, scraper.status_path) 

170 soup = BeautifulSoup(response.text, "lxml") 

171 return resolve_html_value(soup, scraper.assignee_path), resolve_html_value(soup, scraper.status_path)