feat: TEA (Transparency Exchange API) integration

pyproject.toml

@@ -35,6 +35,7 @@ dependencies = [
     "conan>=2.0,<3",
     "questionary>=2.0.1,<3",
     "pipdeptree>=2.0.0",
+    "libtea>=0.4.0,<1",
 ]
 
 [project.urls]
@@ -106,7 +107,6 @@ exclude = [
     "docs/",
 ]
 
-
 [tool.ruff]
 line-length = 120
 target-version = "py310"

sbomify_action/_enrichment/enricher.py

@@ -32,6 +32,7 @@
     PURLSource,
     PyPISource,
     RepologySource,
+    TeaSource,
 )
 
 
@@ -57,16 +58,20 @@ def create_default_registry() -> SourceRegistry:
 
     Tier 2 - Primary Aggregators (40-49):
     - DepsDevSource (40) - Google Open Source Insights
+    - TeaSource (43) - TEA server (auto-discovers from PURL type, TEA_BASE_URL overrides)
     - EcosystemsSource (45) - ecosyste.ms multi-ecosystem aggregator
 
     Tier 3 - Fallback Sources (70-99):
     - PURLSource (70) - local PURL extraction for OS packages (no API)
     - ClearlyDefinedSource (75) - license and attribution data
     - RepologySource (90) - cross-distro metadata (rate-limited)
 
-    Sources are queried sequentially in priority order. If a source returns
-    all required NTIA fields (description, licenses, supplier), subsequent
-    sources are skipped.
+    Sources are queried sequentially in priority order with two-phase early exit:
+    1. If NTIA fields (description, licenses, supplier) AND all CLE fields
+       (release_date, eos, eol) are filled, remaining sources are skipped.
+    2. If only NTIA fields are filled but CLE is missing, non-CLE sources
+       are skipped while lifecycle-capable sources (provides_cle=True)
+       continue to run.
 
     Returns:
         Configured SourceRegistry
@@ -80,6 +85,7 @@ def create_default_registry() -> SourceRegistry:
     registry.register(ConanSource())
     registry.register(DebianSource())
     registry.register(DepsDevSource())
+    registry.register(TeaSource())
     registry.register(EcosystemsSource())
     registry.register(PURLSource())
     registry.register(ClearlyDefinedSource())
@@ -271,6 +277,7 @@ def clear_all_caches() -> None:
     from .sources.pubdev import clear_cache as clear_pubdev
     from .sources.pypi import clear_cache as clear_pypi
     from .sources.repology import clear_cache as clear_repology
+    from .sources.tea import clear_cache as clear_tea
 
     clear_license_db()
     clear_lifecycle()
@@ -280,6 +287,7 @@ def clear_all_caches() -> None:
     clear_conan()
     clear_debian()
     clear_depsdev()
+    clear_tea()
     clear_ecosystems()
     clear_clearlydefined()
     clear_repology()

sbomify_action/_enrichment/metadata.py

@@ -129,4 +129,5 @@ def has_data(self) -> bool:
             or self.maintainer_name
             or self.cle_eos
             or self.cle_eol
+            or self.cle_release_date
         )

sbomify_action/_enrichment/registry.py

@@ -1,6 +1,6 @@
 """Source registry for managing data source plugins."""
 
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 import requests
 from packageurl import PackageURL
@@ -33,7 +33,7 @@ class SourceRegistry:
 
     def __init__(self) -> None:
         """Initialize an empty registry."""
-        self._sources: List[DataSource] = []
+        self._sources: list[DataSource] = []
 
     def register(self, source: DataSource) -> None:
         """
@@ -47,7 +47,7 @@ def register(self, source: DataSource) -> None:
         self._sources.append(source)
         logger.debug(f"Registered data source: {source.name} (priority={source.priority})")
 
-    def get_sources_for(self, purl: PackageURL) -> List[DataSource]:
+    def get_sources_for(self, purl: PackageURL) -> list[DataSource]:
         """
         Get all applicable sources for a PURL, sorted by priority.
 
@@ -66,13 +66,16 @@ def fetch_metadata(
         purl: PackageURL,
         session: requests.Session,
         merge_results: bool = True,
-    ) -> Optional[NormalizedMetadata]:
+    ) -> NormalizedMetadata | None:
         """
         Fetch metadata using the priority chain of sources.
 
-        Tries sources in priority order, stopping early when we have
-        sufficient data (description, licenses, supplier). Only continues
-        to lower-priority sources if critical fields are missing.
+        Tries sources in priority order with two-phase early exit:
+        1. If NTIA fields (description, licenses, supplier) AND all CLE
+           fields (release_date, eos, eol) are filled, stop entirely.
+        2. If only NTIA fields are filled but CLE is missing, skip
+           non-CLE sources and continue only with lifecycle-capable
+           sources (those declaring ``provides_cle = True``).
 
         Args:
             purl: Parsed PackageURL object
@@ -87,13 +90,22 @@ def fetch_metadata(
             logger.debug(f"No sources available for PURL type: {purl.type}")
             return None
 
-        result: Optional[NormalizedMetadata] = None
+        result: NormalizedMetadata | None = None
 
         for source in sources:
-            # Stop early if we already have all core NTIA fields
+            # Two-phase early exit: (1) if NTIA + all CLE fields are filled, stop entirely;
+            # (2) if only NTIA is filled but CLE is missing, skip non-CLE sources and
+            # continue only with lifecycle-capable sources (those declaring provides_cle=True).
             if result and result.description and result.licenses and result.supplier:
-                logger.debug(f"Skipping {source.name} - already have sufficient data for {purl.name}")
-                break
+                if result.cle_release_date and result.cle_eos and result.cle_eol:
+                    logger.debug(f"Skipping {source.name} - already have sufficient data for {purl.name}")
+                    break
+                # NTIA complete but CLE missing — only continue with lifecycle-capable sources.
+                # Sources opt in by declaring `provides_cle = True` (not in the Protocol;
+                # checked via getattr so non-CLE sources need no changes).
+                if not getattr(source, "provides_cle", False):
+                    logger.debug(f"Skipping {source.name} - NTIA complete, not a CLE provider for {purl.name}")
+                    continue
 
             try:
                 metadata = source.fetch(purl, session)
@@ -114,7 +126,7 @@ def fetch_metadata(
 
         return result
 
-    def list_sources(self) -> List[Dict[str, Any]]:
+    def list_sources(self) -> list[dict[str, Any]]:
         """
         List all registered sources with their priorities.
 

sbomify_action/_enrichment/sources/__init__.py

@@ -12,6 +12,7 @@
 from .purl import PURLSource
 from .pypi import PyPISource
 from .repology import RepologySource
+from .tea import TeaSource
 
 __all__ = [
     "ClearlyDefinedSource",
@@ -26,4 +27,5 @@
     "PURLSource",
     "PyPISource",
     "RepologySource",
+    "TeaSource",
 ]

sbomify_action/_enrichment/sources/lifecycle.py

@@ -63,6 +63,10 @@ class LifecycleSource:
     def name(self) -> str:
         return "sbomify-lifecycle-db"
 
+    @property
+    def provides_cle(self) -> bool:
+        return True
+
     @property
     def priority(self) -> int:
         # Priority 5: Very high - local data with no API calls