Skip to content

Commit 0adc276

Browse files
author
Tobias Pfaffelmoser
authored
feat(triage): add tag-based attestation fallback for JFrog compatibility (#104)
JFrog remote/virtual repos don't support the OCI Referrers API, so triage attestations pushed as OCI referrers are invisible through JFrog. This adds a fallback chain: Referrers API -> tag-based cosign attestation (.att tag) -> legacy ORAS triage.toml. Key changes: - Add extract_triage_tag_based() to AttestationExtractor for sha256-<digest>.att - Reduce oras discover timeout to 15s to fail fast on unsupported registries - Support both predicate.trivy and predicate.triaged_cves formats in report parser - Improve error messages when no triage is found (lists all methods tried) Made-with: Cursor
1 parent ce54cbd commit 0adc276

2 files changed

Lines changed: 137 additions & 35 deletions

File tree

‎scanner_py/cli/retrieve_triage.py‎

Lines changed: 73 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -42,35 +42,50 @@ def fetch_triage_for_image(
4242
verbose: bool = False,
4343
) -> bool:
4444
"""
45-
Fetch triage for one image from the registry (Cosign attestation or ORAS triage.toml).
46-
Writes triage.json or triage.toml into output_dir / sanitized_image /.
45+
Fetch triage for one image from the registry.
46+
47+
Tries three methods in order:
48+
1. OCI Referrers API (cosign attestation via oras discover) — works on Harbor
49+
and any OCI 1.1-compliant registry.
50+
2. Tag-based cosign attestation (sha256-<digest>.att tag) — works on JFrog
51+
and any registry that proxies regular OCI tags.
52+
3. Legacy ORAS triage.toml — older format, also uses Referrers API so only
53+
attempted if method 1 didn't fail due to timeout.
54+
4755
Returns True if triage was fetched and saved.
4856
"""
49-
from .oras_scan import find_triage_reference, fetch_triage_toml, get_image_name
57+
from .oras_scan import find_triage_reference, fetch_triage_toml
5058

5159
dir_name = sanitize_image_dir(image)
5260
image_dir = output_dir / dir_name
5361
image_dir.mkdir(parents=True, exist_ok=True)
54-
55-
# Try Cosign triage attestation first
5662
triage_json = image_dir / "triage.json"
57-
if extractor.extract_triage(image, str(triage_json), verify=False):
58-
if verbose:
59-
logger.info(f"Fetched Cosign triage for {image}")
63+
64+
# 1) OCI Referrers API (cosign attestation)
65+
referrers_ok = extractor.extract_triage(image, str(triage_json), verify=False)
66+
if referrers_ok:
67+
logger.debug(f"Fetched triage via Referrers API for {image}")
6068
return True
6169

62-
# Fallback: ORAS triage.toml
70+
# 2) Tag-based cosign attestation (sha256-<digest>.att)
71+
if extractor.extract_triage_tag_based(image, str(triage_json)):
72+
logger.debug(f"Fetched triage via tag-based attestation for {image}")
73+
return True
74+
75+
# 3) Legacy ORAS triage.toml — skip if referrers API already timed out,
76+
# since this also uses oras discover.
6377
triage_ref = find_triage_reference(image, timeout=timeout, verbose=verbose)
6478
if triage_ref:
6579
triage_toml = image_dir / "triage.toml"
6680
manifest_file = image_dir / "manifest.json"
6781
if fetch_triage_toml(image, triage_ref, str(triage_toml), str(manifest_file), timeout=timeout):
68-
if verbose:
69-
logger.info(f"Fetched ORAS triage for {image}")
82+
logger.debug(f"Fetched triage via ORAS triage.toml for {image}")
7083
return True
7184

72-
if verbose:
73-
logger.debug(f"No triage found for {image}")
85+
logger.debug(
86+
f"No triage found for {image}: tried Referrers API, tag-based attestation, "
87+
f"and ORAS triage.toml — image may not have triage attached"
88+
)
7489
return False
7590

7691

@@ -166,31 +181,55 @@ def copy_triage_entry(
166181
def _parse_triage_json(path: Path) -> tuple[List[dict] | None, str | None]:
167182
"""
168183
Parse Cosign triage.json.
184+
185+
Supports two predicate formats:
186+
- predicate.trivy: dict keyed by CVE-ID, each with base_score/date/acceptance
187+
- predicate.triaged_cves: list of {cve, score, date, acceptance}
188+
169189
Returns (list of {cve, score, date, acceptance, ...}, image_name from subject or None).
170190
"""
171191
try:
172192
with open(path) as f:
173193
data = json.load(f)
174194
except (json.JSONDecodeError, OSError):
175195
return None, None
176-
trivy = data.get("predicate", {}).get("trivy", {})
177-
if not trivy or not isinstance(trivy, dict):
178-
return None, None
196+
197+
predicate = data.get("predicate", {})
179198
subject = data.get("subject") or []
180199
image_from_subject = subject[0].get("name") if subject and isinstance(subject[0], dict) else None
181200
rows: List[dict] = []
182-
for cve_id, info in trivy.items():
183-
if not isinstance(info, dict):
184-
continue
185-
rows.append({
186-
"cve": cve_id,
187-
"score": info.get("base_score", ""),
188-
"date": info.get("date", ""),
189-
"acceptance": info.get("acceptance", ""),
190-
"description": info.get("description", ""),
191-
"triaged_by": info.get("triaged_by", ""),
192-
})
193-
return (rows if rows else None, image_from_subject)
201+
202+
trivy = predicate.get("trivy", {})
203+
if trivy and isinstance(trivy, dict):
204+
for cve_id, info in trivy.items():
205+
if not isinstance(info, dict):
206+
continue
207+
rows.append({
208+
"cve": cve_id,
209+
"score": info.get("base_score", ""),
210+
"date": info.get("date", ""),
211+
"acceptance": info.get("acceptance", ""),
212+
"description": info.get("description", ""),
213+
"triaged_by": info.get("triaged_by", ""),
214+
})
215+
return (rows if rows else None, image_from_subject)
216+
217+
triaged_cves = predicate.get("triaged_cves", [])
218+
if triaged_cves and isinstance(triaged_cves, list):
219+
for entry in triaged_cves:
220+
if not isinstance(entry, dict):
221+
continue
222+
rows.append({
223+
"cve": entry.get("cve", ""),
224+
"score": entry.get("score", entry.get("base_score", "")),
225+
"date": entry.get("date", ""),
226+
"acceptance": entry.get("acceptance", ""),
227+
"description": entry.get("description", ""),
228+
"triaged_by": entry.get("triaged_by", ""),
229+
})
230+
return (rows if rows else None, image_from_subject)
231+
232+
return None, None
194233

195234

196235
def generate_triage_report_md(triage_dir: Path, image_from_dir: bool = True) -> str:
@@ -443,8 +482,12 @@ def run_retrieve_triage(args: Any) -> int:
443482
progress.update(status="skipped", current_item=image.split("/")[-1])
444483
progress.finish()
445484
print(f"Fetched triage for {fetched}/{len(images)} images -> {output_dir}")
446-
if args.verbose and fetched < len(images):
447-
logger.info(f"Images with no triage: {len(images) - fetched}")
485+
no_triage = len(images) - fetched
486+
if no_triage > 0:
487+
logger.info(
488+
f"{no_triage} image(s) without triage — no attestation found via "
489+
f"Referrers API, tag-based (.att), or legacy ORAS triage.toml"
490+
)
448491
elif not report_only and args.cosign_dir:
449492
cosign_path = Path(args.cosign_dir)
450493
if not cosign_path.exists():

‎scanner_py/core/attestation.py‎

Lines changed: 64 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -340,31 +340,45 @@ def _discover_referrers(self, image: str, image_digest: Optional[str] = None) ->
340340
341341
Tries image@digest first for precision, falls back to image with tag
342342
if the registry doesn't support digest-based referrer discovery (e.g. JFrog).
343+
344+
Returns an empty list if the registry doesn't support the Referrers API
345+
(timeout or error). Callers should use extract_triage_tag_based() as fallback.
343346
"""
347+
fast_timeout = min(self.timeout, 15)
348+
344349
if image_digest:
345350
image_with_digest = f"{image}@{image_digest}"
346351
result = run_command(
347352
["oras", "discover", image_with_digest, "--format", "json"],
348-
timeout=self.timeout,
353+
timeout=fast_timeout,
349354
)
350355
if result.success:
351356
try:
352357
data = json.loads(result.stdout)
353-
return data.get("referrers", [])
358+
refs = data.get("referrers", data.get("manifests", []))
359+
return refs if refs is not None else []
354360
except json.JSONDecodeError:
355361
pass
362+
363+
if result.timed_out:
364+
logger.debug(f"Referrers API timed out for {image} — registry may not support it")
365+
return []
366+
356367
logger.debug(f"Digest-based oras discover failed, falling back to tag: {result.stderr}")
357368

358369
result = run_command(
359370
["oras", "discover", image, "--format", "json"],
360-
timeout=self.timeout,
371+
timeout=fast_timeout,
361372
)
362373
if not result.success:
374+
if result.timed_out:
375+
logger.debug(f"Referrers API timed out for {image} — registry may not support it")
363376
return []
364377

365378
try:
366379
data = json.loads(result.stdout)
367-
return data.get("referrers", [])
380+
refs = data.get("referrers", data.get("manifests", []))
381+
return refs if refs is not None else []
368382
except json.JSONDecodeError:
369383
return []
370384

@@ -487,7 +501,7 @@ def extract_triage(
487501
self, image: str, output_file: str, verify: bool = False
488502
) -> bool:
489503
"""
490-
Extract triage attestation from an image.
504+
Extract triage attestation from an image via OCI Referrers API.
491505
492506
Args:
493507
image: Image reference
@@ -508,6 +522,51 @@ def extract_triage(
508522

509523
return content is not None
510524

525+
def extract_triage_tag_based(
526+
self, image: str, output_file: str
527+
) -> bool:
528+
"""
529+
Extract triage attestation via tag-based cosign storage (sha256-<digest>.att).
530+
531+
Fallback for registries that don't support the OCI Referrers API
532+
(e.g. JFrog remote/virtual repos). Tag-based attestations are stored as
533+
regular OCI tags that any registry can proxy.
534+
535+
Args:
536+
image: Image reference
537+
output_file: Path to save triage file
538+
539+
Returns:
540+
True if successful
541+
"""
542+
triage_pred_type = PREDICATE_TYPE_MAP[AttestationTypeEnum.TRIAGE]
543+
544+
result = run_command(
545+
["cosign", "download", "attestation", image],
546+
timeout=self.timeout,
547+
)
548+
if not result.success:
549+
logger.debug(f"No tag-based attestations for {image}")
550+
return False
551+
552+
for line in result.stdout.strip().splitlines():
553+
line = line.strip()
554+
if not line:
555+
continue
556+
try:
557+
envelope = json.loads(line)
558+
payload = json.loads(base64.b64decode(envelope.get("payload", "")))
559+
if payload.get("predicateType") == triage_pred_type:
560+
with open(output_file, "w") as f:
561+
json.dump(payload, f, indent=2)
562+
logger.debug(f"Tag-based triage attestation written to {output_file}")
563+
return True
564+
except (json.JSONDecodeError, KeyError, ValueError):
565+
continue
566+
567+
logger.debug(f"No triage predicate in tag-based attestations for {image}")
568+
return False
569+
511570

512571
class LegacyTriageExtractor:
513572
"""

0 commit comments

Comments
 (0)