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} 

29 

30 

31async def fetch_osv_advisories( 

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

33) -> list[SecurityAdvisory]: 

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

35 ecosystem = OSV_ECOSYSTEMS.get(info.registry) 

36 if not ecosystem or not info.latest_version: 

37 return [] 

38 

39 advisories: list[SecurityAdvisory] = [] 

40 page_token: str | None = None 

41 for _ in range(_OSV_MAX_PAGES): 

42 query: dict[str, Any] = { 

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

44 "version": info.latest_version, 

45 } 

46 if page_token: 

47 query["page_token"] = page_token 

48 

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

50 response.raise_for_status() 

51 payload = response.json() 

52 

53 advisories.extend( 

54 advisory 

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

56 if isinstance(raw, dict) 

57 and not raw.get("withdrawn") 

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

59 ) 

60 

61 page_token = payload.get("next_page_token") 

62 if not page_token: 

63 break 

64 

65 return advisories 

66 

67 

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

69 advisory_id = raw.get("id") 

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

71 return None 

72 

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

74 if not isinstance(affected, list): 

75 affected = [] 

76 fixed_versions = sorted( 

77 { 

78 event["fixed"] 

79 for item in affected 

80 if isinstance(item, dict) 

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

82 if isinstance(range_item, dict) 

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

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

85 } 

86 ) 

87 references = [ 

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

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

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

91 ] 

92 return SecurityAdvisory( 

93 id=advisory_id, 

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

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

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

97 severity=_severity(raw, affected), 

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

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

100 fixed_versions=fixed_versions, 

101 references=references, 

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

103 ) 

104 

105 

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

107 for item in affected: 

108 if not isinstance(item, dict): 

109 continue 

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

111 if isinstance(source, dict): 

112 value = source.get("severity") 

113 if isinstance(value, str) and value: 

114 return value.upper() 

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

116 if isinstance(item, dict): 

117 score = item.get("score") 

118 if isinstance(score, str) and score: 

119 return score 

120 return "" 

121 

122 

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

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

125 return None 

126 with suppress(ValueError, TypeError): 

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

128 return None