[RORDEV-1813] fix: better handling of ROR metadata embedded into header value (#1176)

Author: coutoPL

core/src/main/scala/tech/beshu/ror/accesscontrol/domain/http.scala

@@ -69,40 +69,36 @@ object Header {
 
   def apply(nameAndValue: (NonEmptyString, NonEmptyString)): Header = new Header(Name(nameAndValue._1), nameAndValue._2)
 
-  def fromRawHeaders(headers: Map[String, List[String]]): Set[Header] = {
-    val (authorizationHeaders, otherHeaders) =
+  def fromRawHeaders(headers: Map[String, List[String]]): Either[AuthorizationValueError, Set[Header]] = {
+    val (authorizationHeaders, nonAuthorizationHeaders) =
       headers
         .map { case (name, values) => (name, values.toCovariantSet) }
-        .flatMap { case (name, values) =>
-          for {
-            nonEmptyName <- NonEmptyString.unapply(name)
-            nonEmptyValues <- NonEmptyList.fromList(values.toList.flatMap(NonEmptyString.unapply))
-          } yield (Header.Name(nonEmptyName), nonEmptyValues)
-        }
-        .toSeq
-        .partition { case (name, _) => name === Header.Name.authorization }
-    val headersFromAuthorizationHeaderValues = authorizationHeaders
-      .flatMap { case (_, values) =>
-        val headersFromAuthorizationHeaderValues = values
-          .map(fromAuthorizationValue)
-          .toList
-          .map(_.map(_.toList))
-          .sequence
-          .map(_.flatten)
-        headersFromAuthorizationHeaderValues match {
-          case Left(error) => throw new IllegalArgumentException(error.show)
-          case Right(v) => v
-        }
+        .flatMap { case (name, values) => createHeadersFrom(name, values) }
+        .partition(h => h.name === Header.Name.authorization)
+    val headersFromAuthorizationHeaderValues =
+      authorizationHeaders.toList
+        .map(header => fromAuthorizationValue(header.value))
+        .sequence
+        .map(_.flatMap(_.toList))
+
+    headersFromAuthorizationHeaderValues
+      .map { authHeaderBasedExtractedHeaders =>
+        val restOfHeadersNames = nonAuthorizationHeaders.map(_.name).toCovariantSet
+        val filteredAuthHeaderBasedExtractedHeaders = authHeaderBasedExtractedHeaders
+          .filter { header => !restOfHeadersNames.contains(header.name) }
+        (nonAuthorizationHeaders ++ filteredAuthHeaderBasedExtractedHeaders).toCovariantSet
       }
-      .toCovariantSet
-    val restOfHeaders = otherHeaders
-      .flatMap { case (name, values) => values.map(new Header(name, _)).toList }
-      .toCovariantSet
-    val restOfHeaderNames = restOfHeaders.map(_.name)
-    restOfHeaders ++ headersFromAuthorizationHeaderValues.filter { header => !restOfHeaderNames.contains(header.name) }
   }
 
-  def fromAuthorizationValue(value: NonEmptyString): Either[AuthorizationValueError, NonEmptyList[Header]] = {
+  private def createHeadersFrom(name: String, values: Iterable[String]) = {
+    val value = for {
+      nonEmptyName <- NonEmptyString.unapply(name)
+      nonEmptyValues <- NonEmptyList.fromList(values.toList.flatMap(NonEmptyString.unapply))
+    } yield nonEmptyValues.toList.map(value => new Header(Header.Name(nonEmptyName), value))
+    value.toList.flatten
+  }
+
+  private def fromAuthorizationValue(value: NonEmptyString): Either[AuthorizationValueError, NonEmptyList[Header]] = {
     value.value.splitBy("ror_metadata=") match {
       case (_, None) =>
         Right(NonEmptyList.one(new Header(Name.authorization, value)))
@@ -182,7 +178,7 @@ object Address {
     Hostname.fromString(value).map(Address.Name.apply)
 
   private def parseIpAddress(value: String) =
-    (cutOffZoneIndex _ andThen IpAddress.fromString andThen (_.map(createAddressIp))) (value)
+    (cutOffZoneIndex _ andThen IpAddress.fromString andThen (_.map(createAddressIp)))(value)
 
   private def createAddressIp(ip: IpAddress) =
     Address.Ip(Cidr(ip, 32))
@@ -214,9 +210,9 @@ final case class UriPath private(value: NonEmptyString) {
   def isSqlQueryPath: Boolean = value.value.startsWith("/_sql")
 
   def isXpackSqlQueryPath: Boolean = value.value.startsWith("/_xpack/sql")
-  
+
   def isEsqlQueryPath: Boolean = value.value.startsWith("/_query")
-  
+
   def isAliasesPath: Boolean =
     value.value.startsWith("/_cat/aliases") ||
       value.value.startsWith("/_alias") ||

core/src/main/scala/tech/beshu/ror/implicits.scala

@@ -33,7 +33,6 @@ import tech.beshu.ror.accesscontrol.blocks.definitions.ldap.implementations.Unbo
 import tech.beshu.ror.accesscontrol.blocks.definitions.ldap.implementations.UserGroupsSearchFilterConfig.UserGroupsSearchMode.*
 import tech.beshu.ror.accesscontrol.blocks.definitions.ldap.{Dn, LdapService}
 import tech.beshu.ror.accesscontrol.blocks.metadata.UserMetadata
-import tech.beshu.ror.accesscontrol.blocks.rules.Rule
 import tech.beshu.ror.accesscontrol.blocks.rules.Rule.{RuleName, RuleResult}
 import tech.beshu.ror.accesscontrol.blocks.rules.elasticsearch.{ActionsRule, FieldsRule, FilterRule, ResponseFieldsRule}
 import tech.beshu.ror.accesscontrol.blocks.rules.kibana.*
@@ -45,7 +44,6 @@ import tech.beshu.ror.accesscontrol.blocks.variables.startup.StartupResolvableVa
 import tech.beshu.ror.accesscontrol.blocks.variables.transformation.domain.*
 import tech.beshu.ror.accesscontrol.domain.*
 import tech.beshu.ror.accesscontrol.domain.AccessRequirement.{MustBeAbsent, MustBePresent}
-import tech.beshu.ror.accesscontrol.domain.Address.Ip
 import tech.beshu.ror.accesscontrol.domain.ClusterIndexName.Remote.ClusterName
 import tech.beshu.ror.accesscontrol.domain.FieldLevelSecurity.Strategy
 import tech.beshu.ror.accesscontrol.domain.GroupIdLike.GroupId
@@ -143,6 +141,7 @@ trait LogsShowInstances
   implicit val proxyAuthNameShow: Show[ProxyAuth.Name] = Show.show(_.value)
 
   implicit def requestedIndexShow[T <: ClusterIndexName : Show]: Show[RequestedIndex[T]] = Show(_.name.show)
+
   implicit val clusterIndexNameShow: Show[ClusterIndexName] = Show.show(_.stringify)
   implicit val localClusterIndexNameShow: Show[ClusterIndexName.Local] = Show.show(_.stringify)
   implicit val remoteClusterIndexNameShow: Show[ClusterIndexName.Remote] = Show.show(_.stringify)
@@ -381,19 +380,25 @@ trait LogsShowInstances
     showNamedIterable(name, option.toList)
   }
 
-  implicit val authorizationValueErrorShow: Show[AuthorizationValueError] = Show.show {
+  val authorizationValueErrorWithDetailsShow: Show[AuthorizationValueError] = Show.show {
     case AuthorizationValueError.EmptyAuthorizationValue => "Empty authorization value"
     case AuthorizationValueError.InvalidHeaderFormat(value) => s"Unexpected header format in ror_metadata: [${value.show}]"
     case AuthorizationValueError.RorMetadataInvalidFormat(value, message) => s"Invalid format of ror_metadata: [${value.show}], reason: [${message.show}]"
   }
 
+  val authorizationValueErrorSanitizedShow: Show[AuthorizationValueError] = Show.show {
+    case AuthorizationValueError.EmptyAuthorizationValue => "Empty authorization value"
+    case AuthorizationValueError.InvalidHeaderFormat(_) => s"Unexpected header format in ror_metadata"
+    case AuthorizationValueError.RorMetadataInvalidFormat(_, message) => s"Invalid format of ror_metadata. Reason: [${message.show}]"
+  }
+
   implicit val unresolvableErrorShow: Show[Unresolvable] = Show.show {
     case Unresolvable.CannotExtractValue(msg) => s"Cannot extract variable value. ${msg.show}"
     case Unresolvable.CannotInstantiateResolvedValue(msg) => s"Extracted value type doesn't fit. ${msg.show}"
   }
 
   implicit def accessShow[T: Show]: Show[AccessRequirement[T]] = Show.show {
     case MustBePresent(value) => value.show
-    case AccessRequirement.MustBeAbsent(value) => s"~${value.show}"
+    case MustBeAbsent(value) => s"~${value.show}"
   }
 }

es67x/src/main/scala/tech/beshu/ror/es/RorRestChannel.scala

@@ -22,38 +22,56 @@ import squants.information.{Bytes, Information}
 import tech.beshu.ror.accesscontrol.domain.{Address, Header, UriPath}
 import tech.beshu.ror.accesscontrol.request.RequestContext.Method
 import tech.beshu.ror.accesscontrol.request.RestRequest
+import tech.beshu.ror.es.utils.ThreadRepo
 import tech.beshu.ror.syntax.*
 import tech.beshu.ror.utils.RefinedUtils.nes
 
 import java.net.{InetSocketAddress, SocketAddress}
 import scala.jdk.CollectionConverters.*
 
-final class RorRestChannel(underlying: EsRestChannel)
+object RorRestChannel {
+  def from(esRestChannel: EsRestChannel): Either[Header.AuthorizationValueError, RorRestChannel] = {
+    RorRestRequest
+      .from(esRestChannel.request())
+      .map(new RorRestChannel(esRestChannel, _))
+  }
+}
+final class RorRestChannel private(underlying: EsRestChannel, val restRequest: RorRestRequest)
   extends AbstractRestChannel(underlying.request(), true)
     with ResponseFieldsFiltering
     with Logging {
 
-  val restRequest: RorRestRequest = new RorRestRequest(underlying.request())
-
   override def sendResponse(response: EsRestResponse): Unit = {
+    ThreadRepo.removeRestChannel(this)
     underlying.sendResponse(filterRestResponse(response))
   }
 }
 
-final class RorRestRequest(underlying: EsRestRequest) extends RestRequest {
+object RorRestRequest {
+
+  def from(esRestRequest: EsRestRequest): Either[Header.AuthorizationValueError, RorRestRequest] = {
+    headersFrom(esRestRequest).map(new RorRestRequest(esRestRequest, _))
+  }
+
+  private def headersFrom(esRestRequest: EsRestRequest) = {
+    Header.fromRawHeaders(
+      esRestRequest
+        .getHeaders.asScala
+        .view.mapValues(_.asScala.toList)
+        .toMap
+    )
+  }
+}
+final class RorRestRequest private(underlying: EsRestRequest,
+                                   headers: Set[Header]) extends RestRequest {
 
   override lazy val method: Method = Method.fromStringUnsafe(underlying.method().name())
 
   override lazy val path: UriPath = UriPath
     .from(underlying.path())
     .getOrElse(UriPath.from(nes("/")))
 
-  override lazy val allHeaders: Set[Header] = Header.fromRawHeaders(
-    underlying
-      .getHeaders.asScala
-      .view.mapValues(_.asScala.toList)
-      .toMap
-  )
+  override lazy val allHeaders: Set[Header] = headers
 
   override lazy val localAddress: Address =
     createAddressFrom(_.getLocalAddress)

es67x/src/main/scala/tech/beshu/ror/es/handler/request/context/types/MultiGetEsRequestContext.scala

@@ -26,7 +26,6 @@ import org.elasticsearch.threadpool.ThreadPool
 import tech.beshu.ror.accesscontrol.blocks.BlockContext.FilterableMultiRequestBlockContext
 import tech.beshu.ror.accesscontrol.blocks.BlockContext.MultiIndexRequestBlockContext.Indices
 import tech.beshu.ror.accesscontrol.blocks.metadata.UserMetadata
-import tech.beshu.ror.accesscontrol.domain
 import tech.beshu.ror.accesscontrol.domain.*
 import tech.beshu.ror.accesscontrol.domain.DocumentAccessibility.{Accessible, Inaccessible}
 import tech.beshu.ror.accesscontrol.domain.FieldLevelSecurity.RequestFieldsUsage

es67x/src/main/scala/tech/beshu/ror/es/handler/request/context/types/templates/GetTemplatesEsRequestContext.scala

@@ -107,7 +107,7 @@ class GetTemplatesEsRequestContext(actionRequest: GetIndexTemplatesRequest,
 
 private[templates] object GetTemplatesEsRequestContext extends Logging {
 
-  def filter(templates: List[IndexTemplateMetaData],
+  def filter(templates: Iterable[IndexTemplateMetaData],
              usingTemplate: Set[Template] => Set[Template])
             (implicit requestContextId: RequestContext.Id): List[IndexTemplateMetaData] = {
     val templatesMap = templates

es67x/src/main/scala/tech/beshu/ror/es/utils/ChannelInterceptingRestHandlerDecorator.scala

@@ -16,31 +16,50 @@
  */
 package tech.beshu.ror.es.utils
 
+import cats.implicits.*
+import org.apache.logging.log4j.scala.Logging
+import org.elasticsearch.ElasticsearchException
 import org.elasticsearch.client.node.NodeClient
 import org.elasticsearch.common.settings.Settings
+import org.elasticsearch.rest.*
 import org.elasticsearch.rest.action.cat.RestCatAction
-import org.elasticsearch.rest.{RestChannel, RestHandler, RestRequest}
 import org.joor.Reflect.on
+import tech.beshu.ror.accesscontrol.domain.Header.AuthorizationValueError
 import tech.beshu.ror.es.RorRestChannel
 import tech.beshu.ror.es.actions.wrappers._cat.rest.RorWrappedRestCatAction
 import tech.beshu.ror.es.utils.ThreadContextOps.createThreadContextOps
+import tech.beshu.ror.implicits.*
 import tech.beshu.ror.utils.AccessControllerHelper.doPrivileged
 
-import scala.util.Try
+import scala.util.{Failure, Success, Try}
 
 class ChannelInterceptingRestHandlerDecorator private(val underlying: RestHandler,
                                                       settings: Settings)
-  extends RestHandler {
+  extends RestHandler with Logging {
 
   private val wrapped = doPrivileged {
     wrapSomeActions(underlying)
   }
 
   override def handleRequest(request: RestRequest, channel: RestChannel, client: NodeClient): Unit = {
-    val rorRestChannel = new RorRestChannel(channel)
-    ThreadRepo.setRestChannel(rorRestChannel)
-    addXpackUserAuthenticationHeaderForInCaseOfSecurityRequest(request, client)
-    wrapped.handleRequest(request, rorRestChannel, client)
+    Try {
+      RorRestChannel.from(channel) match {
+        case Right(rorRestChannel) =>
+          ThreadRepo.safeSetRestChannel(rorRestChannel) {
+            addXpackUserAuthenticationHeaderForInCaseOfSecurityRequest(request, client)
+            wrapped.handleRequest(request, rorRestChannel, client)
+          }
+        case Left(error) =>
+          logError(error)
+          implicit val show = authorizationValueErrorSanitizedShow
+          channel.sendResponse(new BytesRestResponse(channel, RestStatus.BAD_REQUEST, new ElasticsearchException(error.show)))
+      }
+    } match {
+      case Success(_) =>
+      case Failure(ex) =>
+        logger.error(s"The incoming request handling error:", ex)
+        channel.sendResponse(new BytesRestResponse(channel, RestStatus.INTERNAL_SERVER_ERROR, new ElasticsearchException("ROR internal error")))
+    }
   }
 
   override def canTripCircuitBreaker: Boolean = underlying.canTripCircuitBreaker
@@ -70,14 +89,25 @@ class ChannelInterceptingRestHandlerDecorator private(val underlying: RestHandle
   }
 
   private def addXpackUserAuthenticationHeaderForInCaseOfSecurityRequest(request: RestRequest,
-                                                                       client: NodeClient): Unit = {
+                                                                         client: NodeClient): Unit = {
     if (request.path().contains("/_security") || request.path().contains("/_xpack/security")) {
       client
         .threadPool().getThreadContext
         .addXpackUserAuthenticationHeader(client.getLocalNodeId)
     }
   }
 
+  private def logError(error: AuthorizationValueError): Unit = {
+    {
+      implicit val show = authorizationValueErrorSanitizedShow
+      logger.warn(s"The incoming request was malformed. Cause: ${error.show}")
+    }
+    if (logger.delegate.isDebugEnabled()) {
+      implicit val show = authorizationValueErrorWithDetailsShow
+      logger.debug(s"Malformed request detailed cause: ${error.show}")
+    }
+  }
+
 }
 
 object ChannelInterceptingRestHandlerDecorator {

es67x/src/main/scala/tech/beshu/ror/es/utils/ThreadRepo.scala

@@ -20,11 +20,19 @@ import org.elasticsearch.rest.RestRequest
 import tech.beshu.ror.accesscontrol.domain.UriPath
 import tech.beshu.ror.es.RorRestChannel
 
+import scala.util.{Failure, Success, Try}
+
 object ThreadRepo {
   private val threadLocalChannel = new ThreadLocal[RorRestChannel]
 
-  def setRestChannel(restChannel: RorRestChannel): Unit = {
+  def safeSetRestChannel(restChannel: RorRestChannel)(code: => Unit): Unit = {
     threadLocalChannel.set(restChannel)
+    Try(code) match {
+      case Success(_) =>
+      case Failure(ex) =>
+        removeRestChannel(restChannel)
+        throw ex
+    }
   }
 
   def removeRestChannel(restChannel: RorRestChannel): Unit = {

es70x/src/main/scala/tech/beshu/ror/es/RorRestChannel.scala

@@ -23,38 +23,56 @@ import squants.information.{Bytes, Information}
 import tech.beshu.ror.accesscontrol.domain.{Address, Header, UriPath}
 import tech.beshu.ror.accesscontrol.request.RequestContext.Method
 import tech.beshu.ror.accesscontrol.request.RestRequest
+import tech.beshu.ror.es.utils.ThreadRepo
 import tech.beshu.ror.syntax.*
 import tech.beshu.ror.utils.RefinedUtils.nes
 
 import java.net.InetSocketAddress
 import scala.jdk.CollectionConverters.*
 
-final class RorRestChannel(underlying: EsRestChannel)
+object RorRestChannel {
+  def from(esRestChannel: EsRestChannel): Either[Header.AuthorizationValueError, RorRestChannel] = {
+    RorRestRequest
+      .from(esRestChannel.request())
+      .map(new RorRestChannel(esRestChannel, _))
+  }
+}
+final class RorRestChannel private(underlying: EsRestChannel, val restRequest: RorRestRequest)
   extends AbstractRestChannel(underlying.request(), true)
     with ResponseFieldsFiltering
     with Logging {
 
-  val restRequest: RorRestRequest = new RorRestRequest(underlying.request())
-
   override def sendResponse(response: EsRestResponse): Unit = {
+    ThreadRepo.removeRestChannel(this)
     underlying.sendResponse(filterRestResponse(response))
   }
 }
 
-final class RorRestRequest(underlying: EsRestRequest) extends RestRequest {
+object RorRestRequest {
+
+  def from(esRestRequest: EsRestRequest): Either[Header.AuthorizationValueError, RorRestRequest] = {
+    headersFrom(esRestRequest).map(new RorRestRequest(esRestRequest, _))
+  }
+
+  private def headersFrom(esRestRequest: EsRestRequest) = {
+    Header.fromRawHeaders(
+      esRestRequest
+        .getHeaders.asScala
+        .view.mapValues(_.asScala.toList)
+        .toMap
+    )
+  }
+}
+final class RorRestRequest private(underlying: EsRestRequest,
+                                   headers: Set[Header]) extends RestRequest {
 
   override lazy val method: Method = Method.fromStringUnsafe(underlying.method().name())
 
   override lazy val path: UriPath = UriPath
     .from(underlying.path())
     .getOrElse(UriPath.from(nes("/")))
 
-  override lazy val allHeaders: Set[Header] = Header.fromRawHeaders(
-    underlying
-      .getHeaders.asScala
-      .view.mapValues(_.asScala.toList)
-      .toMap
-  )
+  override lazy val allHeaders: Set[Header] = headers
 
   override lazy val localAddress: Address =
     createAddressFrom(_.getLocalAddress)