Configure set-cookie header

http4s/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/CollectorService.scala

@@ -2,12 +2,15 @@ package com.snowplowanalytics.snowplow.collectors.scalastream
 
 import java.util.UUID
 
+import scala.concurrent.duration._
 import scala.collection.JavaConverters._
 
-import cats.effect.Sync
+import cats.effect.{Clock, Sync}
 import cats.implicits._
 
-import org.http4s.{Request, RequestCookie, Response}
+import org.http4s._
+import org.http4s.headers._
+import org.http4s.implicits._
 import org.http4s.Status._
 
 import org.typelevel.ci._
@@ -81,8 +84,18 @@ class CollectorService[F[_]: Sync](
         contentType,
         headers(request, spAnonymous)
       )
+      now <- Clock[F].realTime
+      setCookie = cookieHeader(
+        headers       = request.headers,
+        cookieConfig  = config.cookieConfig,
+        networkUserId = nuid,
+        doNotTrack    = doNotTrack,
+        spAnonymous   = spAnonymous,
+        now           = now
+      )
+      responseHeaders = Headers(setCookie.toList.map(_.toRaw1))
       _ <- sinkEvent(event, partitionKey)
-    } yield buildHttpResponse
+    } yield buildHttpResponse(responseHeaders)
 
   def determinePath(vendor: String, version: String): String = {
     val original = s"/$vendor/$version"
@@ -122,7 +135,8 @@ class CollectorService[F[_]: Sync](
   }
 
   // TODO: Handle necessary cases to build http response in here
-  def buildHttpResponse: Response[F] = Response(status = Ok)
+  def buildHttpResponse(headers: Headers): Response[F] =
+    Response(status = Ok, headers = headers)
 
   // TODO: Since Remote-Address and Raw-Request-URI is akka-specific headers,
   // they aren't included in here. It might be good to search for counterparts in Http4s.
@@ -152,6 +166,100 @@ class CollectorService[F[_]: Sync](
       _ <- sinks.bad.storeRawEvents(eventSplit.bad, partitionKey)
     } yield ()
 
+  /**
+    * Builds a cookie header with the network user id as value.
+    *
+    * @param cookieConfig  cookie configuration extracted from the collector configuration
+    * @param networkUserId value of the cookie
+    * @param doNotTrack    whether do not track is enabled or not
+    * @return the build cookie wrapped in a header
+    */
+  def cookieHeader(
+    headers: Headers,
+    cookieConfig: Option[CookieConfig],
+    networkUserId: String,
+    doNotTrack: Boolean,
+    spAnonymous: Option[String],
+    now: FiniteDuration
+  ): Option[`Set-Cookie`] =
+    if (doNotTrack) {
+      None
+    } else {
+      spAnonymous match {
+        case Some(_) => None
+        case None =>
+          cookieConfig.map { config =>
+            val responseCookie = ResponseCookie(
+              name    = config.name,
+              content = networkUserId,
+              expires = Some(HttpDate.unsafeFromEpochSecond((now + config.expiration).toSeconds)),
+              domain  = cookieDomain(headers, config.domains, config.fallbackDomain),
+              path    = Some("/"),
+              sameSite = config.sameSite.flatMap {
+                case "Strict" => SameSite.Strict.some
+                case "Lax"    => SameSite.Lax.some
+                case "None"   => SameSite.None.some
+                case _        => None
+              },
+              secure   = config.secure,
+              httpOnly = config.httpOnly
+            )
+            `Set-Cookie`(responseCookie)
+          }
+      }
+    }
+
+  /**
+    * Determines the cookie domain to be used by inspecting the Origin header of the request
+    * and trying to find a match in the list of domains specified in the config file.
+    *
+    * @param headers        The headers from the http request.
+    * @param domains        The list of cookie domains from the configuration.
+    * @param fallbackDomain The fallback domain from the configuration.
+    * @return The domain to be sent back in the response, unless no cookie domains are configured.
+    *         The Origin header may include multiple domains. The first matching domain is returned.
+    *         If no match is found, the fallback domain is used if configured. Otherwise, the cookie domain is not set.
+    */
+  def cookieDomain(
+    headers: Headers,
+    domains: Option[List[String]],
+    fallbackDomain: Option[String]
+  ): Option[String] =
+    (for {
+      domainList <- domains
+      originHosts = extractHosts(headers)
+      domainToUse <- domainList.find(domain => originHosts.exists(validMatch(_, domain)))
+    } yield domainToUse).orElse(fallbackDomain)
+
+  /** Extracts the host names from a list of values in the request's Origin header. */
+  def extractHosts(headers: Headers): List[String] =
+    (for {
+      // We can't use 'headers.get[Origin]' function in here because of the bug
+      // reported here: https://github.com/http4s/http4s/issues/7236
+      // To circumvent the bug, we split the the Origin header value with blank char
+      // and parse items individually.
+      originSplit <- headers.get(ci"Origin").map(_.head.value.split(' '))
+      parsed = originSplit.map(Origin.parse(_).toOption).toList.flatten
+      hosts  = parsed.flatMap(extractHostFromOrigin)
+    } yield hosts).getOrElse(List.empty)
+
+  private def extractHostFromOrigin(originHeader: Origin): List[String] =
+    originHeader match {
+      case Origin.Null            => List.empty
+      case Origin.HostList(hosts) => hosts.map(_.host.value).toList
+    }
+
+  /**
+    * Ensures a match is valid.
+    * We only want matches where:
+    * a.) the Origin host is exactly equal to the cookie domain from the config
+    * b.) the Origin host is a subdomain of the cookie domain from the config.
+    * But we want to avoid cases where the cookie domain from the config is randomly
+    * a substring of the Origin host, without any connection between them.
+    */
+  def validMatch(host: String, domain: String): Boolean =
+    host == domain || host.endsWith("." + domain)
+
   /**
     * Gets the IP from a RemoteAddress. If ipAsPartitionKey is false, a UUID will be generated.
     *

http4s/src/main/scala/com.snowplowanalytics.snowplow.collectors.scalastream/model.scala

@@ -1,5 +1,7 @@
 package com.snowplowanalytics.snowplow.collectors.scalastream
 
+import scala.concurrent.duration._
+
 import io.circe.Json
 
 object model {
@@ -27,7 +29,21 @@ object model {
     */
   final case class SplitBatchResult(goodBatches: List[List[Json]], failedBigEvents: List[Json])
 
-  final case class CollectorConfig(
-    paths: Map[String, String]
+  final case class CookieConfig(
+    enabled: Boolean,
+    name: String,
+    expiration: FiniteDuration,
+    domains: Option[List[String]],
+    fallbackDomain: Option[String],
+    secure: Boolean,
+    httpOnly: Boolean,
+    sameSite: Option[String]
   )
+
+  final case class CollectorConfig(
+    paths: Map[String, String],
+    cookie: CookieConfig
+  ) {
+    val cookieConfig = if (cookie.enabled) Some(cookie) else None
+  }
 }