Implementation of ROLIE feeds

csaf-retrieval/src/main/kotlin/io/github/csaf/sbom/retrieval/CsafLoader.kt

@@ -19,6 +19,7 @@ package io.github.csaf.sbom.retrieval
 import io.github.csaf.sbom.schema.generated.Aggregator
 import io.github.csaf.sbom.schema.generated.Csaf
 import io.github.csaf.sbom.schema.generated.Provider
+import io.github.csaf.sbom.schema.generated.ROLIEFeed
 import io.ktor.client.*
 import io.ktor.client.call.*
 import io.ktor.client.engine.*
@@ -96,6 +97,19 @@ class CsafLoader(engine: HttpClientEngine = Java.create()) {
     suspend fun fetchDocument(url: String, ctx: RetrievalContext): Result<Csaf> =
         Result.of { get<Csaf>(url, ctx.responseCallback()).also(ctx.jsonCallback()) }
 
+    /**
+     * Fetch and parse a ROLE feed from a given URL.
+     *
+     * @param url the URL where the ROLIE feed is found
+     * @param responseCallback An optional callback to further evaluate the [HttpResponse].
+     * @return The resulting [ROLIEFeed], wrapped in a [Result] monad, if successful. A failed
+     *   [Result] wrapping the thrown [Throwable] in case of an error.
+     */
+    suspend fun fetchROLIEFeed(
+        url: String,
+        responseCallback: ((HttpResponse) -> Unit)? = null
+    ): Result<ROLIEFeed> = Result.of { get(url, responseCallback ?: {}) }
+
     /**
      * Fetch an arbitrary URL's content as plain text [String], falling back to UTF-8 if no charset
      * is provided.

csaf-retrieval/src/main/kotlin/io/github/csaf/sbom/retrieval/RetrievedProvider.kt

@@ -21,11 +21,14 @@ import io.github.csaf.sbom.retrieval.roles.CSAFProviderRole
 import io.github.csaf.sbom.retrieval.roles.CSAFPublisherRole
 import io.github.csaf.sbom.retrieval.roles.CSAFTrustedProviderRole
 import io.github.csaf.sbom.schema.generated.Provider
+import io.github.csaf.sbom.schema.generated.Provider.Feed
+import io.github.csaf.sbom.schema.generated.ROLIEFeed
 import java.util.*
 import java.util.concurrent.CompletableFuture
 import java.util.stream.Stream
 import java.util.stream.StreamSupport
 import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.ProducerScope
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.produce
 import kotlinx.coroutines.channels.toList
@@ -36,6 +39,7 @@ import kotlinx.coroutines.future.future
  * "trusted provider"), including its metadata (in the form of [Provider]) as well as functionality
  * to retrieve its documents.
  */
+@OptIn(ExperimentalCoroutinesApi::class)
 class RetrievedProvider(val json: Provider) : Validatable {
 
     /**
@@ -51,21 +55,22 @@ class RetrievedProvider(val json: Provider) : Validatable {
             }
 
     /**
-     * This function fetches all directory indices referenced by this provider.
+     * This function fetches all directory indices referenced by this provider and sends them to a
+     * [ReceiveChannel].
      *
      * @param loader The instance of [CsafLoader] used for fetching of online resources.
      * @param channelCapacity The capacity of the channels used to buffer parallel fetches. Defaults
      *   to [DEFAULT_CHANNEL_CAPACITY].
-     * @return The fetched [Result]s, representing index contents or fetch errors.
+     * @return A [ReceiveChannel] containing the fetched [Result]s, representing index contents or
+     *   fetch errors.
      */
-    @OptIn(ExperimentalCoroutinesApi::class)
     fun fetchDocumentIndices(
         loader: CsafLoader = lazyLoader,
         channelCapacity: Int = DEFAULT_CHANNEL_CAPACITY
     ): ReceiveChannel<Pair<String, Result<String>>> {
-        @Suppress("SimpleRedundantLet")
         val directoryUrls =
             (json.distributions ?: emptySet()).mapNotNull { distribution ->
+                @Suppress("SimpleRedundantLet")
                 distribution.directory_url?.let { it.toString().trimEnd('/') }
             }
         // This channel collects up to `channelCapacity` directory indices concurrently.
@@ -83,6 +88,37 @@ class RetrievedProvider(val json: Provider) : Validatable {
         }
     }
 
+    /**
+     * This function fetches all ROLIE feeds referenced by this provider and sends them to a
+     * [ReceiveChannel].
+     *
+     * @param loader The instance of [CsafLoader] used for fetching of online resources.
+     * @param channelCapacity The capacity of the channels used to buffer parallel fetches. Defaults
+     *   to [DEFAULT_CHANNEL_CAPACITY].
+     * @return A [ReceiveChannel] containing the fetched [Result]s, representing ROLIE feeds'
+     *   contents (as [ROLIEFeed]) or fetch errors.
+     */
+    fun fetchRolieFeeds(
+        loader: CsafLoader = lazyLoader,
+        channelCapacity: Int = DEFAULT_CHANNEL_CAPACITY
+    ): ReceiveChannel<Pair<Feed, Result<ROLIEFeed>>> {
+        val feeds = json.distributions?.mapNotNull { it.rolie }?.flatMap { it.feeds } ?: listOf()
+
+        // This channel collects up to `channelCapacity` feeds concurrently.
+        val rolieChannel =
+            ioScope.produce(capacity = channelCapacity) {
+                for (feed in feeds) {
+                    send(feed to async { loader.fetchROLIEFeed(feed.url.toString()) })
+                }
+            }
+        // This terminal channel is a simple "rendezvous channel" for awaiting the Results.
+        return ioScope.produce {
+            for ((feed, feedDeferred) in rolieChannel) {
+                send(feed to feedDeferred.await())
+            }
+        }
+    }
+
     /**
      * This function sums up the expected number of [RetrievedDocument]s that will be fetched from
      * this Provider.
@@ -92,7 +128,6 @@ class RetrievedProvider(val json: Provider) : Validatable {
      *   to [DEFAULT_CHANNEL_CAPACITY].
      * @return The expected number of [RetrievedDocument]s provided.
      */
-    @OptIn(ExperimentalCoroutinesApi::class)
     suspend fun countExpectedDocuments(
         loader: CsafLoader = lazyLoader,
         channelCapacity: Int = DEFAULT_CHANNEL_CAPACITY
@@ -124,36 +159,14 @@ class RetrievedProvider(val json: Provider) : Validatable {
         channelCapacity: Int = DEFAULT_CHANNEL_CAPACITY
     ): ReceiveChannel<Result<RetrievedDocument>> {
         val indexChannel = fetchDocumentIndices(loader, channelCapacity)
+        val rolieFeedsChannel = fetchRolieFeeds(loader, channelCapacity)
+
         // This second channel collects up to `channelCapacity` Results concurrently, which
         // represent CSAF Documents or errors from fetching or validation.
         val documentJobChannel =
             ioScope.produce<Deferred<Result<RetrievedDocument>>>(capacity = channelCapacity) {
-                for ((directoryUrl, indexResult) in indexChannel) {
-                    indexResult.fold(
-                        { index ->
-                            index.lines().map { line ->
-                                send(
-                                    async {
-                                        val csafUrl = "$directoryUrl/$line"
-                                        RetrievedDocument.from(csafUrl, loader, role)
-                                    }
-                                )
-                            }
-                        },
-                        { e ->
-                            send(
-                                async {
-                                    Result.failure(
-                                        Exception(
-                                            "Failed to fetch index.txt from directory at $directoryUrl",
-                                            e
-                                        )
-                                    )
-                                }
-                            )
-                        }
-                    )
-                }
+                fetchDocumentsFromIndices(indexChannel, loader)
+                fetchDocumentsFromRolieFeeds(rolieFeedsChannel, loader)
             }
         // This terminal channel is a simple "rendezvous channel" for awaiting the Results.
         return ioScope.produce {
@@ -163,6 +176,70 @@ class RetrievedProvider(val json: Provider) : Validatable {
         }
     }
 
+    /** Populates this [ProducerScope] with the contents of the [indexChannel]. */
+    private suspend fun ProducerScope<Deferred<Result<RetrievedDocument>>>
+        .fetchDocumentsFromIndices(
+        indexChannel: ReceiveChannel<Pair<String, Result<String>>>,
+        loader: CsafLoader
+    ) {
+        for ((directoryUrl, indexResult) in indexChannel) {
+            indexResult.fold(
+                { index ->
+                    index.lines().map { line ->
+                        send(
+                            async {
+                                val csafUrl = "$directoryUrl/$line"
+                                RetrievedDocument.from(csafUrl, loader, role)
+                            }
+                        )
+                    }
+                },
+                { e ->
+                    send(
+                        async {
+                            Result.failure(
+                                Exception(
+                                    "Failed to fetch index.txt from directory at $directoryUrl",
+                                    e
+                                )
+                            )
+                        }
+                    )
+                }
+            )
+        }
+    }
+
+    /** Populates this [ProducerScope] with the contents of the [rolieChannel]. */
+    private suspend fun ProducerScope<Deferred<Result<RetrievedDocument>>>
+        .fetchDocumentsFromRolieFeeds(
+        rolieChannel: ReceiveChannel<Pair<Feed, Result<ROLIEFeed>>>,
+        loader: CsafLoader
+    ) {
+        for ((feed, rolieResult) in rolieChannel) {
+            rolieResult.fold(
+                { rolie ->
+                    rolie.feed.entry.map { entry ->
+                        send(
+                            async {
+                                RetrievedDocument.from(entry.content.src.toString(), loader, role)
+                            }
+                        )
+                    }
+                },
+                { e ->
+                    send(
+                        async {
+                            Result.failure(
+                                Exception("Failed to fetch feeds from directory at ${feed.url}", e)
+                            )
+                        }
+                    )
+                }
+            )
+        }
+    }
+
     /**
      * This method provides the [Result]s of [fetchDocuments] as a Java [Stream] for usage in
      * non-Kotlin environments.

csaf-retrieval/src/test/java/io/github/csaf/sbom/retrieval/RetrievedProviderJavaTest.java

@@ -52,9 +52,9 @@ public void testRetrievedProviderJava() throws InterruptedException, ExecutionEx
         final var documentResultsExplicit = providerExplicit.streamDocuments(loader, DEFAULT_CHANNEL_CAPACITY).toList();
         final var documentResultsExplicitSlow = providerExplicit.streamDocuments(loader, 1).toList();
         assertEquals(
-                4,
+                5,
                 documentResults.size(),
-                "Expected exactly 4 results: One document, two document errors, one index error"
+                "Expected exactly 5 results: Two documents, two document errors, one index error"
         );
         assertEquals(
                 documentResults.size(),

csaf-retrieval/src/test/kotlin/io/github/csaf/sbom/retrieval/CsafLoaderTest.kt

@@ -102,4 +102,13 @@ class CsafLoaderTest {
             }
         assertFalse { result.isSuccess }
     }
+
+    @Test
+    fun testFetchROLIEFeed() = runTest {
+        val result =
+            loader.fetchROLIEFeed("does-not-really-exist.json") {
+                assertSame(HttpStatusCode.NotFound, it.status)
+            }
+        assertFalse { result.isSuccess }
+    }
 }

csaf-retrieval/src/test/kotlin/io/github/csaf/sbom/retrieval/RetrievedProviderTest.kt

@@ -70,15 +70,23 @@ class RetrievedProviderTest {
         assertFalse(documentIndexResults[1].second.isSuccess)
     }
 
+    @Test
+    fun testFetchRolieFeeds() = runTest {
+        val provider = RetrievedProvider.from("example.com").getOrThrow()
+        val rolieFeedsResults = provider.fetchRolieFeeds().toList()
+        assertEquals(1, rolieFeedsResults.size, "Expected exactly 1 result: One parsed ROLIE feed")
+        assertTrue(rolieFeedsResults[0].second.isSuccess)
+    }
+
     private suspend fun providerTest(domain: String) {
         val provider = RetrievedProvider.from(domain).getOrThrow()
         val expectedDocumentCount = provider.countExpectedDocuments()
         assertEquals(3, expectedDocumentCount, "Expected 3 documents")
         val documentResults = provider.fetchDocuments().toList()
         assertEquals(
-            4,
+            5,
             documentResults.size,
-            "Expected exactly 4 results: One document, two document errors, one index error"
+            "Expected exactly 5 results: Two documents, two document errors, one index error"
         )
         // Check some random property on successful document
         assertEquals(

csaf-retrieval/src/test/resources/example.com/.well-known/csaf/feed-tlp-white.json

@@ -0,0 +1,52 @@
+{
+  "feed": {
+    "id": "example-csaf-feed-tlp-white",
+    "title": "Example CSAF feed (TLP:WHITE)",
+    "link": [
+      {
+        "rel": "self",
+        "href": "https://example.com/.well-known/csaf/feed-tlp-white.json"
+      }
+    ],
+    "category": [
+      {
+        "scheme": "urn:ietf:params:rolie:category:information-type",
+        "term": "csaf"
+      }
+    ],
+    "updated": "2021-01-01T12:00:00.000Z",
+    "entry": [
+      {
+        "id": "2020-ESA-001",
+        "title": "Example Security Advisory 001",
+        "link": [
+          {
+            "rel": "self",
+            "href": "https://example.com/directory/2022/bsi-2022-0001.json"
+          },
+          {
+            "rel": "hash",
+            "href": "https://example.com/directory/2022/bsi-2022-0001.json.sha512"
+          },
+          {
+            "rel": "signature",
+            "href": "https://example.com/directory/2022/bsi-2022-0001.json.asc"
+          }
+        ],
+        "published": "2021-01-01T11:00:00.000Z",
+        "updated": "2021-01-01T12:00:00.000Z",
+        "summary": {
+          "content": "Vulnerabilities fixed in ABC 0.0.1"
+        },
+        "content": {
+          "type": "application/json",
+          "src": "https://example.com/directory/2022/bsi-2022-0001.json"
+        },
+        "format": {
+          "schema": "https://docs.oasis-open.org/csaf/csaf/v2.0/csaf_json_schema.json",
+          "version": "2.0"
+        }
+      }
+    ]
+  }
+}