8722405814

Committed 17 Apr 2024 01:02PM UTC coverage: 31.005% (-0.6%) from 31.633%

Build Type

Committed by

Commit Message

chore: resolve conflict

Signed-off-by: Pat Losoponkul <pat.losoponkul@iohk.io>

Pull Request Pull Request #966: feat: key management for Ed25519 and X25519

Run Details

109 of 386 new or added lines in 22 files covered. (28.24%)

386 existing lines in 101 files now uncovered.

4478 of 14443 relevant lines covered (31.0%)

0.31 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

36.21

/prism-agent/service/wallet-api/src/main/scala/io/iohk/atala/agent/walletapi/vault/VaultClient.scala

package io.iohk.atala.agent.walletapi.vault

import io.github.jopenlibs.vault.Vault
import io.github.jopenlibs.vault.VaultConfig
import io.github.jopenlibs.vault.VaultException
import io.github.jopenlibs.vault.api.Logical
import io.github.jopenlibs.vault.api.LogicalUtilities
import io.github.jopenlibs.vault.response.LogicalResponse
import zio.*
import zio.http.*
import zio.json.*

import java.nio.charset.StandardCharsets
import scala.jdk.CollectionConverters.*

trait VaultKVClient {
  def get[T: KVCodec](path: String): Task[Option[T]]
  def set[T: KVCodec](path: String, data: T, customMetadata: Map[String, String] = Map.empty): Task[Unit]
}

class VaultKVClientImpl(vaultRef: Ref[(Vault, VaultConfig)], client: Client) extends VaultKVClient {
  import VaultKVClientImpl.*

  override def get[T: KVCodec](path: String): Task[Option[T]] = {
    for {
      vault <- vaultRef.get.map(_._1)
      maybeData <- ZIO
        .attemptBlocking(
          vault
            .logical()
            .read(path)
        )
        .handleVaultErrorOpt("Error reading a secret from Vault.")
        .map(_.map(_.getData().asScala.toMap))
      decodedData <- maybeData.fold(ZIO.none)(data => ZIO.fromTry(summon[KVCodec[T]].decode(data)).asSome)
    } yield decodedData
  }

  override def set[T: KVCodec](path: String, data: T, customMetadata: Map[String, String]): Task[Unit] = {
    val kv = summon[KVCodec[T]].encode(data)
    for {
      vaultWithConfig <- vaultRef.get
      (vault, vaultConfig) = vaultWithConfig
      _ <- ZIO
        .attemptBlocking {
          vault
            .logical()
            .write(path, kv.asJava)
        }
        .handleVaultError("Error writing a secret to Vault.")
      _ <- ExtendedLogical(vaultConfig, client)
        .writeMetadata(path, customMetadata)
        .when(customMetadata.nonEmpty)
    } yield ()
  }
}

object VaultKVClientImpl {

  private final case class TokenRefreshState(leaseDuration: Duration)

  def fromToken(address: String, token: String): RIO[Client, VaultKVClient] =
    for {
      client <- ZIO.service[Client]
      vault <- createVaultInstance(address, Some(token))
      vaultRef <- Ref.make(vault)
    } yield VaultKVClientImpl(vaultRef, client)

  def fromAppRole(address: String, roleId: String, secretId: String): RIO[Client, VaultKVClient] =
    for {
      client <- ZIO.service[Client]
      vault <- createVaultInstance(address)
      vaultRef <- vaultTokenRefreshLogic(vault._1, address, roleId, secretId)
    } yield VaultKVClientImpl(vaultRef, client)

  private def vaultTokenRefreshLogic(
      authVault: Vault,
      address: String,
      roleId: String,
      secretId: String,
      tokenRefreshBuffer: Duration = 15.seconds,
      retrySchedule: Schedule[Any, Any, Any] = Schedule.spaced(5.second) && Schedule.recurs(10)
  ): Task[Ref[(Vault, VaultConfig)]] = {
    val getToken = ZIO
      .attempt {
        val authResponse = authVault.auth().loginByAppRole(roleId, secretId)
        val ttlSecond = authResponse.getAuthLeaseDuration()
        val token = authResponse.getAuthClientToken()
        (token, ttlSecond)
      }
      .retry(retrySchedule)

    for {
      tokenWithTtl <- getToken
      (token, ttlSecond) = tokenWithTtl
      vaultWithToken <- createVaultInstance(address, Some(token))
      vaultRef <- Ref.make(vaultWithToken)
      _ <- ZIO
        .iterate(TokenRefreshState(ttlSecond.seconds))(_ => true) { state =>
          val durationUntilRefresh = state.leaseDuration.minus(tokenRefreshBuffer).max(1.second)
          for {
            _ <- ZIO.sleep(durationUntilRefresh)
            tokenWithTtl <- getToken
            (token, ttlSecond) = tokenWithTtl
            vaultWithToken <- createVaultInstance(address, Some(token))
            _ <- vaultRef.set(vaultWithToken)
          } yield state.copy(leaseDuration = ttlSecond.seconds)
        }
        .fork
    } yield vaultRef
  }

  private def createVaultInstance(address: String, token: Option[String] = None): Task[(Vault, VaultConfig)] =
    ZIO.attempt {
      val configBuilder = VaultConfig()
        .engineVersion(2)
        .address(address)
      token.foreach(configBuilder.token)
      val config = configBuilder.build()
      Vault.create(config) -> config
    }

  extension [R](resp: RIO[R, LogicalResponse]) {

    /** Handle non 200 Vault response as error and 404 as optioanl */
    def handleVaultErrorOpt(message: String): RIO[R, Option[LogicalResponse]] = {
      resp
        .flatMap { resp =>
          val status = resp.getRestResponse().getStatus()
          val bytes = resp.getRestResponse().getBody()
          val body = new String(bytes, StandardCharsets.UTF_8)
          status match {
            case 200 => ZIO.some(resp)
            case 404 => ZIO.none
            case _ =>
              ZIO
                .fail(Exception(s"$message - Got response status code $status, expected 200"))
                .tapError(_ => ZIO.logError(s"$message - Response status: $status. Response body: $body"))
          }
        }
    }

    /** Handle non 200 Vault response as error */
    def handleVaultError(message: String): RIO[R, LogicalResponse] = {
      resp
        .flatMap { resp =>
          val status = resp.getRestResponse().getStatus()
          val bytes = resp.getRestResponse().getBody()
          val body = new String(bytes, StandardCharsets.UTF_8)
          status match {
            case 200 => ZIO.succeed(resp)
            case _ =>
              ZIO
                .fail(Exception(s"$message - Got response status code $status, expected 200"))
                .tapError(_ => ZIO.logError(s"$message - Response status: $status. Response body: $body"))
          }
        }
    }
  }

  private final case class WriteMetadataRequest(
      custom_metadata: Map[String, String]
  )

  private object WriteMetadataRequest {
    given JsonEncoder[WriteMetadataRequest] = JsonEncoder.derived
    given JsonDecoder[WriteMetadataRequest] = JsonDecoder.derived
  }

  private class ExtendedLogical(config: VaultConfig, client: Client) extends Logical(config) {
    // based on https://github.com/jopenlibs/vault-java-driver/blob/e49312a8cbcd14b260dacb2822c19223feb1b7af/src/main/java/io/github/jopenlibs/vault/api/Logical.java#L275
    def writeMetadata(path: String, metadata: Map[String, String]): Task[Unit] = {
      val pathSegments = path.split("/")
      val pathDepth = config.getPrefixPathDepth()
      val adjustedPath = LogicalUtilities.addQualifierToPath(pathSegments.toSeq.asJava, pathDepth, "metadata")
      val url = config.getAddress() + "/v1/" + adjustedPath

      val baseHeaders = Headers(
        Header.ContentType(MediaType.application.json),
        Header.Custom("X-Vault-Request", "true"),
      )
      val additionalHeaders = Headers(
        Seq(
          Option(config.getToken()).map(Header.Custom("X-Vault-Token", _)),
          Option(config.getNameSpace()).map(Header.Custom("X-Vault-Namespace", _)),
        ).flatten
      )

      for {
        url <- ZIO.fromEither(URL.decode(url)).orDie
        request =
          Request(
            url = url,
            method = Method.POST,
            headers = baseHeaders ++ additionalHeaders,
            body = Body.fromString(WriteMetadataRequest(custom_metadata = metadata).toJson)
          )
        _ <- ZIO
          .scoped(client.request(request))
          .timeoutFail(new RuntimeException("Client request timed out"))(5.seconds)
          .flatMap { resp =>
            if (resp.status.isSuccess) ZIO.unit
            else {
              resp.body
                .asString(StandardCharsets.UTF_8)
                .flatMap { body =>
                  ZIO.fail(
                    VaultException(
                      s"Expecting HTTP status 2xx, but instead receiving ${resp.status.code}.\n Response body: ${body}",
                      resp.status.code
                    )
                  )
                }
            }
          }
          .retry {
            val maxRetry = Option(config.getMaxRetries()).getOrElse(0)
            val retryIntervalMillis = Option(config.getRetryIntervalMilliseconds()).getOrElse(0)
            Schedule.recurs(maxRetry) && Schedule.spaced(retryIntervalMillis.millis)
          }
      } yield ()
    }
  }
}

1	package io.iohk.atala.agent.walletapi.vault
2
3	import io.github.jopenlibs.vault.Vault
4	import io.github.jopenlibs.vault.VaultConfig
5	import io.github.jopenlibs.vault.VaultException
6	import io.github.jopenlibs.vault.api.Logical
7	import io.github.jopenlibs.vault.api.LogicalUtilities
8	import io.github.jopenlibs.vault.response.LogicalResponse
9	import zio.*
10	import zio.http.*
11	import zio.json.*
12
13	import java.nio.charset.StandardCharsets
14	import scala.jdk.CollectionConverters.*
15
16	trait VaultKVClient {
17	def get[T: KVCodec](path: String): Task[Option[T]]	1✔
18	def set[T: KVCodec](path: String, data: T, customMetadata: Map[String, String] = Map.empty): Task[Unit]
19	}
20
21	class VaultKVClientImpl(vaultRef: Ref[(Vault, VaultConfig)], client: Client) extends VaultKVClient {
22	import VaultKVClientImpl.*
23		1✔
24	override def get[T: KVCodec](path: String): Task[Option[T]] = {	1✔
25	for {	×
26	vault <- vaultRef.get.map(_._1)	×
27	maybeData <- ZIO
28	.attemptBlocking(
29	vault	×
30	.logical()	×
31	.read(path)
32	)	×
33	.handleVaultErrorOpt("Error reading a secret from Vault.")	×
34	.map(_.map(_.getData().asScala.toMap))	×
35	decodedData <- maybeData.fold(ZIO.none)(data => ZIO.fromTry(summon[KVCodec[T]].decode(data)).asSome)
36	} yield decodedData
37	}
38		1✔
39	override def set[T: KVCodec](path: String, data: T, customMetadata: Map[String, String]): Task[Unit] = {	1✔
40	val kv = summon[KVCodec[T]].encode(data)	1✔
UNCOV 41	for {	×
42	vaultWithConfig <- vaultRef.get	×
43	(vault, vaultConfig) = vaultWithConfig	1✔
44	_ <- ZIO
45	.attemptBlocking {
46	vault	×
47	.logical()	×
48	.write(path, kv.asJava)
49	}	×
50	.handleVaultError("Error writing a secret to Vault.")	1✔
51	_ <- ExtendedLogical(vaultConfig, client)	×
52	.writeMetadata(path, customMetadata)	×
53	.when(customMetadata.nonEmpty)
54	} yield ()
55	}
56	}
57
58	object VaultKVClientImpl {
59
60	private final case class TokenRefreshState(leaseDuration: Duration)
61		1✔
62	def fromToken(address: String, token: String): RIO[Client, VaultKVClient] =	1✔
63	for {	×
64	client <- ZIO.service[Client]	×
65	vault <- createVaultInstance(address, Some(token))	×
66	vaultRef <- Ref.make(vault)	1✔
67	} yield VaultKVClientImpl(vaultRef, client)
68		×
69	def fromAppRole(address: String, roleId: String, secretId: String): RIO[Client, VaultKVClient] =	×
70	for {	×
71	client <- ZIO.service[Client]	×
72	vault <- createVaultInstance(address)	×
73	vaultRef <- vaultTokenRefreshLogic(vault._1, address, roleId, secretId)	×
74	} yield VaultKVClientImpl(vaultRef, client)
75		×
76	private def vaultTokenRefreshLogic(
77	authVault: Vault,
78	address: String,
79	roleId: String,
80	secretId: String,	×
81	tokenRefreshBuffer: Duration = 15.seconds,	×
82	retrySchedule: Schedule[Any, Any, Any] = Schedule.spaced(5.second) && Schedule.recurs(10)
83	): Task[Ref[(Vault, VaultConfig)]] = {	×
84	val getToken = ZIO
85	.attempt {	×
86	val authResponse = authVault.auth().loginByAppRole(roleId, secretId)	×
87	val ttlSecond = authResponse.getAuthLeaseDuration()	×
88	val token = authResponse.getAuthClientToken()	×
89	(token, ttlSecond)
90	}
91	.retry(retrySchedule)
92		×
93	for {	×
94	tokenWithTtl <- getToken	×
95	(token, ttlSecond) = tokenWithTtl	×
96	vaultWithToken <- createVaultInstance(address, Some(token))	×
97	vaultRef <- Ref.make(vaultWithToken)	×
98	_ <- ZIO	×
99	.iterate(TokenRefreshState(ttlSecond.seconds))(_ => true) { state =>	×
100	val durationUntilRefresh = state.leaseDuration.minus(tokenRefreshBuffer).max(1.second)	×
101	for {	×
102	_ <- ZIO.sleep(durationUntilRefresh)	×
103	tokenWithTtl <- getToken	×
104	(token, ttlSecond) = tokenWithTtl	×
105	vaultWithToken <- createVaultInstance(address, Some(token))	×
106	_ <- vaultRef.set(vaultWithToken)	×
107	} yield state.copy(leaseDuration = ttlSecond.seconds)
108	}
109	.fork
110	} yield vaultRef
111	}
112		1✔
113	private def createVaultInstance(address: String, token: Option[String] = None): Task[(Vault, VaultConfig)] =	1✔
114	ZIO.attempt {	1✔
115	val configBuilder = VaultConfig()	1✔
116	.engineVersion(2)	1✔
117	.address(address)	1✔
118	token.foreach(configBuilder.token)	1✔
119	val config = configBuilder.build()	1✔
120	Vault.create(config) -> config
121	}
122
123	extension [R](resp: RIO[R, LogicalResponse]) {
124
125	/** Handle non 200 Vault response as error and 404 as optioanl */	1✔
126	def handleVaultErrorOpt(message: String): RIO[R, Option[LogicalResponse]] = {	1✔
127	resp
128	.flatMap { resp =>	1✔
129	val status = resp.getRestResponse().getStatus()	1✔
130	val bytes = resp.getRestResponse().getBody()	1✔
131	val body = new String(bytes, StandardCharsets.UTF_8)
132	status match {	1✔
133	case 200 => ZIO.some(resp)	1✔
134	case 404 => ZIO.none	×
135	case _ =>	×
136	ZIO	×
137	.fail(Exception(s"$message - Got response status code $status, expected 200"))	×
138	.tapError(_ => ZIO.logError(s"$message - Response status: $status. Response body: $body"))
139	}
140	}
141	}
142
143	/** Handle non 200 Vault response as error */	1✔
144	def handleVaultError(message: String): RIO[R, LogicalResponse] = {	1✔
145	resp
146	.flatMap { resp =>	1✔
147	val status = resp.getRestResponse().getStatus()	1✔
148	val bytes = resp.getRestResponse().getBody()	1✔
149	val body = new String(bytes, StandardCharsets.UTF_8)
150	status match {	1✔
151	case 200 => ZIO.succeed(resp)	×
152	case _ =>	×
153	ZIO	×
154	.fail(Exception(s"$message - Got response status code $status, expected 200"))	×
155	.tapError(_ => ZIO.logError(s"$message - Response status: $status. Response body: $body"))
156	}
157	}
158	}
159	}
160
161	private final case class WriteMetadataRequest(
162	custom_metadata: Map[String, String]
163	)
164
165	private object WriteMetadataRequest {	×
166	given JsonEncoder[WriteMetadataRequest] = JsonEncoder.derived	×
167	given JsonDecoder[WriteMetadataRequest] = JsonDecoder.derived
168	}
169
170	private class ExtendedLogical(config: VaultConfig, client: Client) extends Logical(config) {
171	// based on https://github.com/jopenlibs/vault-java-driver/blob/e49312a8cbcd14b260dacb2822c19223feb1b7af/src/main/java/io/github/jopenlibs/vault/api/Logical.java#L275	1✔
172	def writeMetadata(path: String, metadata: Map[String, String]): Task[Unit] = {	1✔
173	val pathSegments = path.split("/")	1✔
174	val pathDepth = config.getPrefixPathDepth()	1✔
175	val adjustedPath = LogicalUtilities.addQualifierToPath(pathSegments.toSeq.asJava, pathDepth, "metadata")	1✔
176	val url = config.getAddress() + "/v1/" + adjustedPath
177		1✔
178	val baseHeaders = Headers(	1✔
179	Header.ContentType(MediaType.application.json),
180	Header.Custom("X-Vault-Request", "true"),
181	)	1✔
182	val additionalHeaders = Headers(	×
183	Seq(	×
184	Option(config.getToken()).map(Header.Custom("X-Vault-Token", _)),	×
185	Option(config.getNameSpace()).map(Header.Custom("X-Vault-Namespace", _)),
186	).flatten
187	)
188		1✔
189	for {	×
190	url <- ZIO.fromEither(URL.decode(url)).orDie
191	request =	×
192	Request(
193	url = url,
194	method = Method.POST,	×
195	headers = baseHeaders ++ additionalHeaders,	×
196	body = Body.fromString(WriteMetadataRequest(custom_metadata = metadata).toJson)
197	)	1✔
198	_ <- ZIO	×
199	.scoped(client.request(request))	×
200	.timeoutFail(new RuntimeException("Client request timed out"))(5.seconds)
201	.flatMap { resp =>	×
202	if (resp.status.isSuccess) ZIO.unit	×
203	else {	×
204	resp.body
205	.asString(StandardCharsets.UTF_8)
206	.flatMap { body =>	×
207	ZIO.fail(	×
208	VaultException(	×
209	s"Expecting HTTP status 2xx, but instead receiving ${resp.status.code}.\n Response body: ${body}",
210	resp.status.code
211	)
212	)
213	}
214	}
215	}
216	.retry {	×
217	val maxRetry = Option(config.getMaxRetries()).getOrElse(0)	×
218	val retryIntervalMillis = Option(config.getRetryIntervalMilliseconds()).getOrElse(0)	×
219	Schedule.recurs(maxRetry) && Schedule.spaced(retryIntervalMillis.millis)
220	}
221	} yield ()
222	}
223	}
224	}

input-output-hk / atala-prism-building-blocks / 8722405814

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous