Coverage for src/secchi/security.py: 86%

1"""OSV-backed security advisory lookup for supported package ecosystems.""" 

2 

3from __future__ import annotations 

4 

5from contextlib import suppress 

6from datetime import datetime 

7from typing import Any 

8 

9import httpx 

10 

11from secchi.models import AdvisoryReference, PackageInfo, Registry, SecurityAdvisory 

12 

13OSV_QUERY_URL = "https://api.osv.dev/v1/query" 

14OSV_WEB_URL = "https://osv.dev/vulnerability" 

15 

16# Safety cap on paginated OSV queries. A real package/version is never 

17# expected to need this many pages; it exists purely to bound the loop 

18# against a misbehaving or malicious response that keeps returning a 

19# next_page_token. 

20_OSV_MAX_PAGES = 20 

21 

22OSV_ECOSYSTEMS: dict[Registry, str] = { 

23 Registry.PYPI: "PyPI", 

24 Registry.NPM: "npm", 

25 Registry.CRATES: "crates.io", 

26 Registry.GO: "Go", 

27 Registry.CRAN: "CRAN", 

28 Registry.PUB: "Pub", 

29} 

30 

31 

32async def fetch_osv_advisories( 

33 info: PackageInfo, *, client: httpx.AsyncClient 

34) -> list[SecurityAdvisory]: 

35 """Return advisories affecting ``info.latest_version``.""" 

36 ecosystem = OSV_ECOSYSTEMS.get(info.registry) 

37 if not ecosystem or not info.latest_version: 

38 return [] 

39 

40 advisories: list[SecurityAdvisory] = [] 

41 page_token: str | None = None 

42 for _ in range(_OSV_MAX_PAGES): 

43 query: dict[str, Any] = { 

44 "package": {"name": info.name, "ecosystem": ecosystem}, 

45 "version": info.latest_version, 

46 } 

47 if page_token: 

48 query["page_token"] = page_token 

49 

50 response = await client.post(OSV_QUERY_URL, json=query) 

51 response.raise_for_status() 

52 payload = response.json() 

53 

54 advisories.extend( 

55 advisory 

56 for raw in payload.get("vulns", []) 

57 if isinstance(raw, dict) 

58 and not raw.get("withdrawn") 

59 and (advisory := _parse_advisory(raw)) is not None 

60 ) 

61 

62 page_token = payload.get("next_page_token") 

63 if not page_token: 

64 break 

65 

66 return advisories 

67 

68 

69def _parse_advisory(raw: dict[str, Any]) -> SecurityAdvisory | None: 

70 advisory_id = raw.get("id") 

71 if not isinstance(advisory_id, str) or not advisory_id: 

72 return None 

73 

74 affected = raw.get("affected", []) 

75 if not isinstance(affected, list): 

76 affected = [] 

77 fixed_versions = sorted( 

78 { 

79 event["fixed"] 

80 for item in affected 

81 if isinstance(item, dict) 

82 for range_item in item.get("ranges", []) 

83 if isinstance(range_item, dict) 

84 for event in range_item.get("events", []) 

85 if isinstance(event, dict) and isinstance(event.get("fixed"), str) 

86 } 

87 ) 

88 references = [ 

89 AdvisoryReference(type=item.get("type", ""), url=item["url"]) 

90 for item in raw.get("references", []) 

91 if isinstance(item, dict) and isinstance(item.get("url"), str) 

92 ] 

93 return SecurityAdvisory( 

94 id=advisory_id, 

95 summary=raw.get("summary", "") or "", 

96 details=raw.get("details", "") or "", 

97 aliases=[alias for alias in raw.get("aliases", []) if isinstance(alias, str)], 

98 severity=_severity(raw, affected), 

99 published=_parse_datetime(raw.get("published")), 

100 modified=_parse_datetime(raw.get("modified")), 

101 fixed_versions=fixed_versions, 

102 references=references, 

103 url=f"{OSV_WEB_URL}/{advisory_id}", 

104 ) 

105 

106 

107def _severity(raw: dict[str, Any], affected: list[Any]) -> str: 

108 for item in affected: 

109 if not isinstance(item, dict): 

110 continue 

111 for source in (item.get("ecosystem_specific"), item.get("database_specific")): 

112 if isinstance(source, dict): 

113 value = source.get("severity") 

114 if isinstance(value, str) and value: 

115 return value.upper() 

116 for item in raw.get("severity", []): 

117 if isinstance(item, dict): 

118 score = item.get("score") 

119 if isinstance(score, str) and score: 

120 return score 

121 return "" 

122 

123 

124def _parse_datetime(value: Any) -> datetime | None: 

125 if not isinstance(value, str) or not value: 

126 return None 

127 with suppress(ValueError, TypeError): 

128 return datetime.fromisoformat(value.replace("Z", "+00:00")) 

129 return None