Classify the patch type based on the given URL, commit hash, and patch text. Returns: a base_purl, patch_obj tuple where base_purl is a string PackageURL without version for supported VCS URLs, otherwise `None`. patch_obj is one of: (PackageCommitPatchData for supported VCS URLs with a
(url, commit_hash, patch_text)
| 204 | |
| 205 | |
| 206 | def classify_patch_source(url, commit_hash, patch_text): |
| 207 | """ |
| 208 | Classify the patch type based on the given URL, commit hash, and patch text. |
| 209 | Returns: a base_purl, patch_obj tuple where base_purl is a string PackageURL without version for supported VCS URLs, otherwise `None`. |
| 210 | patch_obj is one of: (PackageCommitPatchData for supported VCS URLs with a commit, |
| 211 | PatchData for raw patch text or non-VCS URLs, ReferenceV2 when unsupported VCS URL is paired with a commit hash) |
| 212 | Returns `None` only when both `url` and `patch_text` are missing. |
| 213 | """ |
| 214 | if not url: |
| 215 | if not patch_text: |
| 216 | return |
| 217 | |
| 218 | return None, [PatchData(patch_text=patch_text)] |
| 219 | |
| 220 | purl = url2purl(url) |
| 221 | if not purl or (purl.type not in VCS_URLS_SUPPORTED_TYPES): |
| 222 | if commit_hash: |
| 223 | if not patch_text: |
| 224 | return None, [ |
| 225 | ReferenceV2( |
| 226 | reference_id=commit_hash, reference_type=AdvisoryReference.COMMIT, url=url |
| 227 | ) |
| 228 | ] |
| 229 | |
| 230 | return None, [ |
| 231 | ReferenceV2( |
| 232 | reference_id=commit_hash, reference_type=AdvisoryReference.COMMIT, url=url |
| 233 | ), |
| 234 | PatchData(patch_url=url, patch_text=patch_text), |
| 235 | ] |
| 236 | |
| 237 | return None, [PatchData(patch_url=url, patch_text=patch_text)] |
| 238 | |
| 239 | if not commit_hash and not purl.version: |
| 240 | return None, [PatchData(patch_url=url, patch_text=patch_text)] |
| 241 | |
| 242 | base_purl = get_core_purl(purl) |
| 243 | base_purl_str = base_purl.to_string() |
| 244 | base_url = purl2url(base_purl_str) |
| 245 | package_commit_hash = purl.version or commit_hash |
| 246 | return base_purl, [ |
| 247 | PackageCommitPatchData( |
| 248 | vcs_url=base_url, commit_hash=package_commit_hash, patch_text=patch_text |
| 249 | ) |
| 250 | ] |
| 251 | |
| 252 | |
| 253 | def insert_advisory(advisory: AdvisoryData, pipeline_id: str, logger: Callable = None): |