Kris Rice's questions

背景：

我有一个名为 的 Actor Peer。在我的应用程序中，Peer表示创建它的服务器与另一台服务器之间的连接。它本质上是一个http-client。

首次运行实例时，它PeerManager会查询数据库以获取服务器条目列表。然后，它会Peer为该服务器创建一个参与者。状态Peer取决于其RegistrationStatus字段（PENDING_OUT、PENDING_IN、REGISTERED，...）

APeer有一个默认的接收函数，如下所示：

  override def receive: Receive = {
    case FirstTick =>
      system.log.info("receive FirstTick")
      timers.startTimerWithFixedDelay(TickKey, Tick, FiniteDuration(2, TimeUnit.SECONDS))
    case Tick =>
      system.log.info("receive Tick")
      // process every state until all are complete
      var statesLeft = false
      
      states = states.sortWith{ (s1, s2) => s1.order > s2.order }

      for (state <- states) {
        if (!state.complete) {
          statesLeft = true
          if (state.waitForPing) {
            context.become(pingService(state))
          } else {
            context.become(state.behavior)
          }
        }
      }

      if (!statesLeft) context.become(pingService(null))
  }

这基本上允许Peer通过状态列表进行转换，执行每个状态的行为直到状态完成。

一组状态可能看起来像这样：

    case PENDING_OUT =>
        Seq(
          new InitState(0, peer),
          new SessionHandshake(1, peer),
          new WaitMessages(2, peer)
        )

其中一个状态的示例：

class WaitAccept(order: Int, peer: Peer)(implicit system: ActorSystem) extends PeerState(order, false, true) {

  override def behavior: Receive = {
    case Tick =>
      system.log.info("Waiting for accept")
      if (complete) peer.getContext.unbecome()

    case rA@RegistrationAccept(serverID, serverIP, _) =>
      system.log.info(s"Processing RegistrationAccept: ${rA.toProtoString}")
      complete = true
  }
  
}

PeerState超类（superclass）：

abstract case class PeerState(var order: Int, var complete: Boolean, var waitForPing: Boolean) {

  def behavior: Receive // <- must be overridden
  
}

此状态使对等方坐等“RegistrationAccept”消息，该消息在用户单击网页上的“接受”时生成。

问题：

某些州（例如，Handshake负责交换加密密钥的州）会向为其生成 的 发送消息http。如果接收消息的实例发现 具有所请求的，则会将该消息转发给 ，如下所示：serverPeerPeerserverIDPeerManagerPeerask

case peerRequest@PeerRequest(serverID, SessionRequest(sessionRequest), _) =>
      // get the relevant peer for the requested serverID
      val peerEntry = peerMap.get(serverID)
      if (peerEntry.isDefined) {
        // ask the peer Actor for a response to the sessionRequest, block and wait for this response, then send it back to the peerManager

        val peerState = queryPeerState(peerEntry.get)    // <-- the issue

        if (peerState.isInstanceOf[SessionHandshake]) {  // <-- the issue
          try {
            sender() ! Await.result((peerEntry.get ? (self, sessionRequest)).mapTo[PeerResponse.Response], timeout.duration)
          } catch {
            case e: TimeoutException =>
              sender() ! PeerResponse.Response.StatusResponse(HydraStatusCodes.PEER_MANAGER_TIMEOUT.getStatusResponse)
          }
        } else {
          sender() ! PeerResponse.Response.StatusResponse(HydraStatusCodes.PEER_NOT_READY.getStatusResponse)
        }
      } else {
        sender() ! PeerResponse.Response.StatusResponse(HydraStatusCodes.PEER_NOT_READY.getStatusResponse)
      }

当其他实例Peer处于接收消息的不正确状态时，就会出现问题。如果发生这种情况，我希望用适当的 做出响应HydraStatusCode。对我来说，最正常的方法是在转发消息之前以某种方式询问Peer它处于什么状态，如果处于错误状态，则返回适当的 StatusCode。

但是，为了实现这一点，我必须QueryState在有Receive行为的每个地方实现案例。这看起来笨重而混乱，作为一名熟练的工程师，我觉得这不是正确的做法。

我希望以某种方式将其实现到PeerState类中作为始终被调用的默认行为QueryState，或者以某种方式允许直接PeerManager调用对Peer对象的引用而不是对 ActorRef 的引用（但我知道这是不可能的）

还有其他方法可以解决这个问题吗？或者混乱的方法可能是最好的？

问题：

在过去一周左右的时间里，我一直在反复尝试配置我的 Scala+AKKA 客户端，以便能够向运行 NGINX 的服务器发送消息。

我不断收到错误： javax.net.ssl.SSLHandshakeException: (certificate_unknown)

设置：

nginx配置：

server {
    listen 443 ssl default_server;
    listen [::]:443 ssl default_server;
    server_name localhost;

    ssl_certificate /home/hydra/.localhost-ssl/localhost.crt;           //<- combined certificates (server_cert + rootCA)
    ssl_certificate_key /home/hydra/.localhost-ssl/localhost-key.key;

    ssl_trusted_certificate /home/hydra/.localhost-ssl/rootCA.crt;      //<- just the rootCA

    index index.html index.htm;
    root /home/hydra/ui/;

    location / {
        try_files $uri.html $uri/index.html
        @public
        @nextjs;
        add_header Cache-Control "public, max-age=3600";
    }

    location @public {
        add_header Cache-Control "public, max-age=3600";
    }

    location /ping {
        proxy_pass http://localhost:8080;
    }

    location @nextjs {
            proxy_pass http://localhost:3000;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection 'upgrade';
            proxy_set_header Host $host;
            proxy_cache_bypass $http_upgrade;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-Proto https;
            proxy_set_header X-Forwarded-For $remote_addr;
            proxy_set_header X-Forwarded-Host $remote_addr;
    }
}

客户要求：

  private val sslContext: SSLContext = SSLManager.getClientSSLContext
  private val connectionContext = ConnectionContext.httpsClient(sslContext)

...

    val request = HttpRequest(method = HttpMethods.GET, uri = s"https://${server.serverIP}/ping")
    http.singleRequest(request, connectionContext).pipeTo(self)

createClientContext：

  def getClientSSLContext: SSLContext = {
    val keyStore = KeyStore.getInstance("JKS")
    keyStore.load(null, null) // Create an empty keystore
    keyStore.setCertificateEntry("rootCA", loadRootCertificate())

    // Set up a TrustManager that trusts the root CA certificate
    val trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm)
    trustManagerFactory.init(keyStore)
    val trustManagers = trustManagerFactory.getTrustManagers

    // Create an SSLContext with the custom TrustManager
    val sslContext = SSLContext.getInstance("TLS")
    sslContext.init(null, trustManagers, new SecureRandom())
    sslContext
  }

写入rootCA文件：

    val rootCA = new StringBuilder()
    rootCA.append("-----BEGIN CERTIFICATE-----\n")
    rootCA.append(Base64.getEncoder.encodeToString(rootCACertificate.getEncoded))
    rootCA.append("\n-----END CERTIFICATE-----")
    writeFile(ROOT_CA_PATH, Seq(rootCA.toString))

如何创建签名的服务器证书：

def createCSR(keyPair: KeyPair, subject: String, keyAlgorithm: String): PKCS10CertificationRequest = {
    val csrGen = new PKCS10CertificationRequestBuilder(new X500Name(subject), SubjectPublicKeyInfo.getInstance(keyPair.getPublic.getEncoded))
    val signer = new JcaContentSignerBuilder("SHA256with" + keyAlgorithm).build(keyPair.getPrivate)
    csrGen.build(signer)
  }

  // Sign the CSR with the root CA's private key to generate a certificate
  def signCertificate(csr: PKCS10CertificationRequest, rootCACertificate: X509Certificate, rootCAPrivateKey: PrivateKey): X509Certificate = {
    val notBefore = new Date()
    val notAfter = new Date(notBefore.getTime + 36500L * 24 * 60 * 60 * 1000) // Valid for 1 year

    val certGen = new X509v3CertificateBuilder(
      new X500Name("CN=Hydra SSL Certificate"),
      new BigInteger(128, new Random()),
      notBefore,
      notAfter,
      csr.getSubject,
      csr.getSubjectPublicKeyInfo
    )

    // Add SubjectAlternativeName (SAN) extension
    val sanNames = Array[GeneralName](
      new GeneralName(GeneralName.iPAddress, SERVER_IP)
    )

    val generalNames = new GeneralNames(sanNames)
    certGen.addExtension(Extension.subjectAlternativeName, false, generalNames)

    // Sign with root CA's private key
    val signer = new JcaContentSignerBuilder("SHA256withRSA").build(rootCAPrivateKey)
    val certificateHolder: X509CertificateHolder = certGen.build(signer)

    // Convert to a JCE certificate
    val converter = new JcaX509CertificateConverter().setProvider("BC")
    converter.getCertificate(certificateHolder)
  }

...

    // Step 1: Load Root CA certificate and private key
    val rootCACertificate = loadRootCertificate()
    val rootCAPrivateKey = loadRootPrivateKey()

    // Step 2: Generate new key pair for SSL certificate
    val keyPair = generateKey("RSA")

    // Step 3: Create CSR (Certificate Signing Request)
    val csr = createCSR(keyPair, s"CN=hydra_server_$SERVER_ID, O=Hydra, C=UK", "RSA")

    // Step 4: Sign the CSR with the Root CA to generate the SSL certificate
    val sslCertificate = signCertificate(csr, rootCACertificate, rootCAPrivateKey)

我尝试过的：

咨询后chatGPT，我尝试了该命令openssl s_client -connect 192.168.0.4:443 -showcerts

该命令的输出可以在这里找到。

这帮助我验证是否nginx确实按正确的顺序发送了整个链（可以通过日期确认 - rootCA（输出中的证书 1）是在 2025 年 7 月 2 日生成的，服务器证书（证书 0）是在今天生成的。

那么我的设置中哪里做错了/遗漏了什么？

我的项目文件夹中有一个rootCA证书resource（及其私钥）。

当软件启动时，它会初始化该SSLManager对象，然后：

• 将 rootCA 加载到x509Certificate
• 加载 rootCA 私钥
• 生成服务器密钥对和证书
• 使用 rootCA 签署证书

我希望将 rootCA 和服务器证书存储到密钥库中，以便以后可以使用它来设置 SSLContext，并实现更安全的存储。

我知道如何将单个密钥对/证书存储到密钥库中，但对于链，我该如何正确地进行这一操作呢？

我推测：

keyStore.setKeyEntry(ALIAS, privateKey, password.toCharArray, Array(rootCA, serverCert))

但是，如果是真的，我应该使用哪个私钥（rootCA 还是服务器）？

如果您能提供完整的例子我将非常感激。

背景

我有一个Actor名为akka 的程序Client，它管理http与https服务器的连接。客户端有许多功能，包括ping服务和tokenFetcher服务。

代表client一个服务器与另一个服务器之间的“连接”。其设计目的是允许一个服务器server与另一个服务器聊天。

其过程client如下：

1. 定期检查ping对方server是否在线
2. 如果对方server在线，则执行auth并获取令牌
3. 如果令牌有效，则清除向我们请求的所有呼叫

我正在努力解决的是第 3 步。我想知道如何在线程 (actor) 之间安全地实现这一点。

我尝试过的：

我正在使用Seq客户端存储的消息，如下所示：

case class SendApiCall(apiCall: ApiCall, route: String, var sent: Boolean = false)

class Client(server: Server) extends Actor {
    private var apiCalls: Seq[SendApiCall] = Seq.empty[SendApiCall]

    ...

    override def receive: Receive = {
        case sendApiCall@SendApiCall(_, _, _) =>
            if (server.onlineStatus == OFFLINE) {
                apiCalls = apiCalls.appended(sendApiCall)
            }
            else {
                sendApiCall(sendApiCall)
            }
        
        case ServerOnline() => // <- this is send to us from the ping service when it first detects the server is online
            
            apiCalls.iterator.foreach( apiCallRequest =>
                if (!apiCallRequest.sent) {
                    sendApiCall(apiCallRequest)
                    apiCallRequest.sent = true
                }
                apiCallRequest
            )

            apiCalls = apiCalls.filterNot(apiCallRequest => apiCallRequest.sent)
    }
}

但是，我相信apiCalls这是mutable这种情况的状态吗？我想知道：

1. 这个线程安全吗？
2. 如果不是线程安全的，我该如何让它变得安全？

我有一个类，它本身就是一个演员：

class Client(server: Server, systemActor: ActorRef) extends Actor {
...
}

我有一个管理已连接客户端的列表。生成客户端 Actor 的 Actor 监视数据库中的“服务器”列表，并为每个服务器生成一个客户端连接，如下所示：

private val clientList = mutable.ArrayBuffer.empty[Client]

class RegistrationWatcherActor(val systemActor: ActorSystem) extends Actor with Timers {

  implicit val system: ActorSystem = context.system

  timers.startSingleTimer(TickKey, FirstTick, FiniteDuration(1, TimeUnit.SECONDS)) // wait 10 seconds to give database time to initiate

  private def checkRegistrations(): Unit = {
    val database = DatabaseUtil.getInstance
    val serversOutgoing: Seq[Server] = database.getAll[Server](classOf[Server])

    for (server <- serversOutgoing) {
      val client = clientList.find{ client => client.server == server }
      if (client.isEmpty) { // client for this server was not found, so create one
        
      }
    }
  }

  def receive: Receive = {
    case FirstTick =>
      timers.startTimerWithFixedDelay(TickKey, Tick, FiniteDuration(1, TimeUnit.SECONDS))
    case Tick =>
      checkRegistrations()
  }
}

问题

在检查中，if (client.isEmpty)我想创建一个实例Client，将其作为参与者生成，并将引用放入列表中，这样它就不会再次被创建。如您所见，我确定客户端是否已创建的方法是将其serverID与数据库中存储的客户端进行检查。

理想情况下，我想这样做：

val client = new Client(server, systemActor)
systemActor.actorOf(Props(client), server.serverID)
clientList.append(client)

请问有什么建议吗？

选择

我意识到我可以列出clientList并将String存储serverID在这里。但是，我最终希望RegistrationWatcher能够控制客户端的各个方面，因此Client最好引用该对象。