當(dāng)前位置：首頁(yè) > 编程资源 > 编程问答 >内容正文

编程问答

Akka-CQRS（16）- gRPC用JWT进行权限管理

發(fā)布時(shí)間：2023/12/10 编程问答 32 豆豆

生活随笔收集整理的這篇文章主要介紹了 Akka-CQRS（16）- gRPC用JWT进行权限管理小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

? ?前面談過(guò)gRPC的SSL/TLS安全機(jī)制，發(fā)現(xiàn)設(shè)置過(guò)程比較復(fù)雜：比如證書(shū)簽名：需要服務(wù)端、客戶端兩頭都設(shè)置等。想想實(shí)際上用JWT會(huì)更加便捷，而且更安全和功能強(qiáng)大，因?yàn)槌齁WT的加密簽名之外還可以把私密的用戶信息放在JWT里加密后在服務(wù)端和客戶端之間傳遞。當(dāng)然，最基本的是通過(guò)對(duì)JWT的驗(yàn)證機(jī)制可以控制客戶端對(duì)某些功能的使用權(quán)限。

通過(guò)JWT實(shí)現(xiàn)gRPC的函數(shù)調(diào)用權(quán)限管理原理其實(shí)很簡(jiǎn)單：客戶端首先從服務(wù)端通過(guò)身份驗(yàn)證獲取JWT，然后在調(diào)用服務(wù)函數(shù)時(shí)把這個(gè)JWT同時(shí)傳給服務(wù)端進(jìn)行權(quán)限驗(yàn)證。客戶端提交身份驗(yàn)證請(qǐng)求返回JWT可以用一個(gè)獨(dú)立的服務(wù)函數(shù)實(shí)現(xiàn)，如下面.proto文件里的GetAuthToken：

message PBPOSCredential {string userid = 1;string password = 2; } message PBPOSToken {string jwt = 1; }service SendCommand {rpc SingleResponse(PBPOSCommand) returns (PBPOSResponse) {};rpc GetTxnItems(PBPOSCommand) returns (stream PBTxnItem) {};rpc GetAuthToken(PBPOSCredential) returns (PBPOSToken) {};}

比較棘手的是如何把JWT從客戶端傳送至服務(wù)端，因?yàn)間RPC基本上騎劫了Request和Response。其中一個(gè)方法是通過(guò)Interceptor來(lái)截取Request的header即metadata。客戶端將JWT寫入metadata，服務(wù)端從metadata讀取JWT。

我們先看看客戶端的Interceptor設(shè)置和使用：

class AuthClientInterceptor(jwt: String) extends ClientInterceptor {def interceptCall[ReqT, RespT](methodDescriptor: MethodDescriptor[ReqT, RespT], callOptions: CallOptions, channel: io.grpc.Channel): ClientCall[ReqT, RespT] =new ForwardingClientCall.SimpleForwardingClientCall[ReqT, RespT](channel.newCall(methodDescriptor, callOptions)) {override def start(responseListener: ClientCall.Listener[RespT], headers: Metadata): Unit = {headers.put(Key.of("jwt", Metadata.ASCII_STRING_MARSHALLER), jwt)super.start(responseListener, headers)}}}...val unsafeChannel = NettyChannelBuilder.forAddress("192.168.0.189",50051).negotiationType(NegotiationType.PLAINTEXT).build()val securedChannel = ClientInterceptors.intercept(unsafeChannel, new AuthClientInterceptor(jwt))val securedClient = SendCommandGrpc.blockingStub(securedChannel)val resp = securedClient.singleResponse(PBPOSCommand())

身份驗(yàn)證請(qǐng)求即JWT獲取是不需要Interceptor的，所以要用沒(méi)有Interceptor的unsafeChannel：?

//build connection channelval unsafeChannel = NettyChannelBuilder.forAddress("192.168.0.189",50051).negotiationType(NegotiationType.PLAINTEXT).build()val authClient = SendCommandGrpc.blockingStub(unsafeChannel)val jwt = authClient.getAuthToken(PBPOSCredential(userid="johnny",password="p4ssw0rd")).jwtprintln(s"got jwt: $jwt")

JWT的構(gòu)建和使用已經(jīng)在前面的幾篇博文里討論過(guò)了：?

package com.datatech.authimport pdi.jwt._ import org.json4s.native.Json import org.json4s._ import org.json4s.jackson.JsonMethods._ import pdi.jwt.algorithms._ import scala.util._object AuthBase {type UserInfo = Map[String, Any]case class AuthBase(algorithm: JwtAlgorithm = JwtAlgorithm.HMD5,secret: String = "OpenSesame",getUserInfo: (String,String) => Option[UserInfo] = null) {ctx =>def withAlgorithm(algo: JwtAlgorithm): AuthBase = ctx.copy(algorithm = algo)def withSecretKey(key: String): AuthBase = ctx.copy(secret = key)def withUserFunc(f: (String, String) => Option[UserInfo]): AuthBase = ctx.copy(getUserInfo = f)def authenticateToken(token: String): Option[String] =algorithm match {case algo: JwtAsymmetricAlgorithm =>Jwt.isValid(token, secret, Seq((algorithm.asInstanceOf[JwtAsymmetricAlgorithm]))) match {case true => Some(token)case _ => None}case _ =>Jwt.isValid(token, secret, Seq((algorithm.asInstanceOf[JwtHmacAlgorithm]))) match {case true => Some(token)case _ => None}}def getUserInfo(token: String): Option[UserInfo] = {algorithm match {case algo: JwtAsymmetricAlgorithm =>Jwt.decodeRawAll(token, secret, Seq(algorithm.asInstanceOf[JwtAsymmetricAlgorithm])) match {case Success(parts) => Some(((parse(parts._2).asInstanceOf[JObject]) \ "userinfo").values.asInstanceOf[UserInfo])case Failure(err) => None}case _ =>Jwt.decodeRawAll(token, secret, Seq(algorithm.asInstanceOf[JwtHmacAlgorithm])) match {case Success(parts) => Some(((parse(parts._2).asInstanceOf[JObject]) \ "userinfo").values.asInstanceOf[UserInfo])case Failure(err) => None}}}def issueJwt(userinfo: UserInfo): String = {val claims = JwtClaim() + Json(DefaultFormats).write(("userinfo", userinfo))Jwt.encode(claims, secret, algorithm)}}}

服務(wù)端Interceptor的構(gòu)建和設(shè)置如下：?

abstract class FutureListener[Q](implicit ec: ExecutionContext) extends Listener[Q] {protected val delegate: Future[Listener[Q]]private val eventually = delegate.foreach _override def onComplete(): Unit = eventually { _.onComplete() }override def onCancel(): Unit = eventually { _.onCancel() }override def onMessage(message: Q): Unit = eventually { _ onMessage message }override def onHalfClose(): Unit = eventually { _.onHalfClose() }override def onReady(): Unit = eventually { _.onReady() }}object Keys {val AUTH_META_KEY: Metadata.Key[String] = of("jwt", Metadata.ASCII_STRING_MARSHALLER)val AUTH_CTX_KEY: Context.Key[String] = key("jwt") }class AuthorizationInterceptor(implicit ec: ExecutionContext) extends ServerInterceptor {override def interceptCall[Q, R](call: ServerCall[Q, R],headers: Metadata,next: ServerCallHandler[Q, R]): Listener[Q] = {val prevCtx = Context.currentval jwt = headers.get(Keys.AUTH_META_KEY)println(s"!!!!!!!!!!! $jwt !!!!!!!!!!")new FutureListener[Q] {protected val delegate = Future {val nextCtx = prevCtx withValue (Keys.AUTH_CTX_KEY, jwt)Contexts.interceptCall(nextCtx, call, headers, next)}}} }trait gRPCServer {def runServer(service: ServerServiceDefinition)(implicit actorSys: ActorSystem): Unit = {import actorSys.dispatcherval server = NettyServerBuilder.forPort(50051).addService(ServerInterceptors.intercept(service,new AuthorizationInterceptor)).build.start// make sure our server is stopped when jvm is shut downRuntime.getRuntime.addShutdownHook(new Thread() {override def run(): Unit = {server.shutdown()server.awaitTermination()}})}}

注意：客戶端上傳的request-header只能在構(gòu)建server時(shí)接觸到，在具體服務(wù)函數(shù)里是無(wú)法調(diào)用request-header的，但gRPC又一個(gè)結(jié)構(gòu)Context可以在兩個(gè)地方都能調(diào)用。所以，我們可以在構(gòu)建server時(shí)把JWT從header搬到Context里。不過(guò)，千萬(wàn)注意這個(gè)Context的讀寫必須在同一個(gè)線程里。在服務(wù)端的Interceptor里我們把JWT從metadata里讀出然后寫入Context。在需要權(quán)限管理的服務(wù)函數(shù)里再?gòu)腃ontext里讀取JWT進(jìn)行驗(yàn)證：?

override def singleResponse(request: PBPOSCommand): Future[PBPOSResponse] = {val jwt = AUTH_CTX_KEY.getprintln(s"***********$jwt**************")val optUserInfo = authenticator.getUserInfo(jwt)val shopid = optUserInfo match {case Some(m) => m("shopid")case None => "invalid token!"}FastFuture.successful(PBPOSResponse(msg=s"shopid:$shopid"))}

JWT的構(gòu)建也是一個(gè)服務(wù)函數(shù)：?

val authenticator = new AuthBase().withAlgorithm(JwtAlgorithm.HS256).withSecretKey("OpenSesame").withUserFunc(getValidUser)override def getAuthToken(request: PBPOSCredential): Future[PBPOSToken] = {getValidUser(request.userid, request.password) match {case Some(userinfo) => FastFuture.successful(PBPOSToken(authenticator.issueJwt(userinfo)))case None => FastFuture.successful(PBPOSToken("Invalid Token!"))}}

還需要一個(gè)模擬的身份驗(yàn)證服務(wù)函數(shù)：?

package com.datatech.authobject MockUserAuthService {type UserInfo = Map[String,Any]case class User(username: String, password: String, userInfo: UserInfo)val validUsers = Seq(User("johnny", "p4ssw0rd",Map("shopid" -> "1101", "userid" -> "101")),User("tiger", "secret", Map("shopid" -> "1101" , "userid" -> "102")))def getValidUser(userid: String, pswd: String): Option[UserInfo] =validUsers.find(user => user.username == userid && user.password == pswd) match {case Some(user) => Some(user.userInfo)case _ => None} }

下面是本次示范的源代碼：

project/plugins.sbt

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.9") addSbtPlugin("net.virtual-void" % "sbt-dependency-graph" % "0.9.2") addSbtPlugin("com.typesafe.sbt" % "sbt-native-packager" % "1.3.15") addSbtPlugin("com.thesamet" % "sbt-protoc" % "0.99.21") addSbtPlugin("net.virtual-void" % "sbt-dependency-graph" % "0.9.2") libraryDependencies += "com.thesamet.scalapb" %% "compilerplugin" % "0.9.0-M6"

build.sbt

name := "grpc-jwt"version := "0.1"version := "0.1"scalaVersion := "2.12.8"scalacOptions += "-Ypartial-unification"val akkaversion = "2.5.23"libraryDependencies := Seq("com.typesafe.akka" %% "akka-cluster-metrics" % akkaversion,"com.typesafe.akka" %% "akka-cluster-sharding" % akkaversion,"com.typesafe.akka" %% "akka-persistence" % akkaversion,"com.lightbend.akka" %% "akka-stream-alpakka-cassandra" % "1.0.1","org.mongodb.scala" %% "mongo-scala-driver" % "2.6.0","com.lightbend.akka" %% "akka-stream-alpakka-mongodb" % "1.0.1","com.typesafe.akka" %% "akka-persistence-query" % akkaversion,"com.typesafe.akka" %% "akka-persistence-cassandra" % "0.97","com.datastax.cassandra" % "cassandra-driver-core" % "3.6.0","com.datastax.cassandra" % "cassandra-driver-extras" % "3.6.0","ch.qos.logback" % "logback-classic" % "1.2.3","io.monix" %% "monix" % "3.0.0-RC2","org.typelevel" %% "cats-core" % "2.0.0-M1","io.grpc" % "grpc-netty" % scalapb.compiler.Version.grpcJavaVersion,"io.netty" % "netty-tcnative-boringssl-static" % "2.0.22.Final","com.thesamet.scalapb" %% "scalapb-runtime" % scalapb.compiler.Version.scalapbVersion % "protobuf","com.thesamet.scalapb" %% "scalapb-runtime-grpc" % scalapb.compiler.Version.scalapbVersion,"com.pauldijou" %% "jwt-core" % "3.0.1","de.heikoseeberger" %% "akka-http-json4s" % "1.22.0","org.json4s" %% "json4s-native" % "3.6.1","com.typesafe.akka" %% "akka-http-spray-json" % "10.1.8","org.json4s" %% "json4s-jackson" % "3.6.7","org.json4s" %% "json4s-ext" % "3.6.7")// (optional) If you need scalapb/scalapb.proto or anything from // google/protobuf/*.proto //libraryDependencies += "com.thesamet.scalapb" %% "scalapb-runtime" % scalapb.compiler.Version.scalapbVersion % "protobuf"PB.targets in Compile := Seq(scalapb.gen() -> (sourceManaged in Compile).value )enablePlugins(JavaAppPackaging)

main/protobuf/posmessages.proto

syntax = "proto3";import "google/protobuf/wrappers.proto"; import "google/protobuf/any.proto"; import "scalapb/scalapb.proto";option (scalapb.options) = {// use a custom Scala package name// package_name: "io.ontherocks.introgrpc.demo"// don't append file name to packageflat_package: true// generate one Scala file for all messages (services still get their own file)single_file: true// add imports to generated file// useful when extending traits or using custom types// import: "io.ontherocks.hellogrpc.RockingMessage"// code to put at the top of generated file// works only with `single_file: true`//preamble: "sealed trait SomeSealedTrait" };package com.datatech.pos.messages;message PBVchState { //單據(jù)狀態(tài)string opr = 1; //收款員int64 jseq = 2; //begin journal sequence for read-side replayint32 num = 3; //當(dāng)前單號(hào)int32 seq = 4; //當(dāng)前序號(hào)bool void = 5; //取消模式bool refd = 6; //退款模式bool susp = 7; //掛單bool canc = 8; //廢單bool due = 9; //當(dāng)前余額string su = 10; //主管編號(hào)string mbr = 11; //會(huì)員號(hào)int32 mode = 12; //當(dāng)前操作流程：0=logOff, 1=LogOn, 2=Payment }message PBTxnItem { //交易記錄string txndate = 1; //交易日期string txntime = 2; //錄入時(shí)間string opr = 3; //操作員int32 num = 4; //銷售單號(hào)int32 seq = 5; //交易序號(hào)int32 txntype = 6; //交易類型int32 salestype = 7; //銷售類型int32 qty = 8; //交易數(shù)量int32 price = 9; //單價(jià)（分）int32 amount = 10; //碼洋（分）int32 disc = 11; //折扣率 (%)int32 dscamt = 12; //折扣額：負(fù)值 net實(shí)洋 = amount + dscamtstring member = 13; //會(huì)員卡號(hào)string code = 14; //編號(hào)（商品、卡號(hào)...）string acct = 15; //賬號(hào)string dpt = 16; //部類 }message PBPOSResponse {int32 sts = 1;string msg = 2;PBVchState voucher = 3;repeated PBTxnItem txnitems = 4;}message PBPOSCommand {string commandname = 1;string delimitedparams = 2; }message PBPOSCredential {string userid = 1;string password = 2; } message PBPOSToken {string jwt = 1; }service SendCommand {rpc SingleResponse(PBPOSCommand) returns (PBPOSResponse) {};rpc GetTxnItems(PBPOSCommand) returns (stream PBTxnItem) {};rpc GetAuthToken(PBPOSCredential) returns (PBPOSToken) {};}

gRPCServer.scala

package com.datatech.grpc.serverimport io.grpc.ServerServiceDefinition import io.grpc.netty.NettyServerBuilder import io.grpc.ServerInterceptors import scala.concurrent._ import io.grpc.Context import io.grpc.Contexts import io.grpc.ServerCall import io.grpc.ServerCallHandler import io.grpc.ServerInterceptor import io.grpc.Metadata import io.grpc.Metadata.Key.of import io.grpc.Context.key import io.grpc.ServerCall.Listener import akka.actor._abstract class FutureListener[Q](implicit ec: ExecutionContext) extends Listener[Q] {protected val delegate: Future[Listener[Q]]private val eventually = delegate.foreach _override def onComplete(): Unit = eventually { _.onComplete() }override def onCancel(): Unit = eventually { _.onCancel() }override def onMessage(message: Q): Unit = eventually { _ onMessage message }override def onHalfClose(): Unit = eventually { _.onHalfClose() }override def onReady(): Unit = eventually { _.onReady() }}object Keys {val AUTH_META_KEY: Metadata.Key[String] = of("jwt", Metadata.ASCII_STRING_MARSHALLER)val AUTH_CTX_KEY: Context.Key[String] = key("jwt") }class AuthorizationInterceptor(implicit ec: ExecutionContext) extends ServerInterceptor {override def interceptCall[Q, R](call: ServerCall[Q, R],headers: Metadata,next: ServerCallHandler[Q, R]): Listener[Q] = {val prevCtx = Context.currentval jwt = headers.get(Keys.AUTH_META_KEY)println(s"!!!!!!!!!!! $jwt !!!!!!!!!!")new FutureListener[Q] {protected val delegate = Future {val nextCtx = prevCtx withValue (Keys.AUTH_CTX_KEY, jwt)Contexts.interceptCall(nextCtx, call, headers, next)}}} }trait gRPCServer {def runServer(service: ServerServiceDefinition)(implicit actorSys: ActorSystem): Unit = {import actorSys.dispatcherval server = NettyServerBuilder.forPort(50051).addService(ServerInterceptors.intercept(service,new AuthorizationInterceptor)).build.start// make sure our server is stopped when jvm is shut downRuntime.getRuntime.addShutdownHook(new Thread() {override def run(): Unit = {server.shutdown()server.awaitTermination()}})}}

POSServices.scala

package com.datatech.pos.service import com.datatech.grpc.server.Keys._ import akka.http.scaladsl.util.FastFuture import com.datatech.pos.messages._ import com.datatech.grpc.server._ import com.datatech.auth.MockUserAuthService._import scala.concurrent.Future import com.datatech.auth.AuthBase._ import pdi.jwt._ import akka.actor._ import io.grpc.stub.StreamObserverobject POSServices extends gRPCServer {type UserInfo = Map[String, Any]class POSServices extends SendCommandGrpc.SendCommand {val authenticator = new AuthBase().withAlgorithm(JwtAlgorithm.HS256).withSecretKey("OpenSesame").withUserFunc(getValidUser)override def getTxnItems(request: PBPOSCommand, responseObserver: StreamObserver[PBTxnItem]): Unit = ???override def singleResponse(request: PBPOSCommand): Future[PBPOSResponse] = {val jwt = AUTH_CTX_KEY.getprintln(s"***********$jwt**************")val optUserInfo = authenticator.getUserInfo(jwt)val shopid = optUserInfo match {case Some(m) => m("shopid")case None => "invalid token!"}FastFuture.successful(PBPOSResponse(msg=s"shopid:$shopid"))}override def getAuthToken(request: PBPOSCredential): Future[PBPOSToken] = {getValidUser(request.userid, request.password) match {case Some(userinfo) => FastFuture.successful(PBPOSToken(authenticator.issueJwt(userinfo)))case None => FastFuture.successful(PBPOSToken("Invalid Token!"))}}}def main(args: Array[String]) = {implicit val system = ActorSystem("grpc-system")val svc = SendCommandGrpc.bindService(new POSServices, system.dispatcher)runServer(svc)} }

AuthBase.scala

POSClient.scala

package com.datatech.pos.clientimport com.datatech.pos.messages.{PBPOSCommand, PBPOSCredential, SendCommandGrpc} import io.grpc.stub.StreamObserver import io.grpc.netty.{ NegotiationType, NettyChannelBuilder} import io.grpc.CallOptions import io.grpc.ClientCall import io.grpc.ClientInterceptor import io.grpc.ForwardingClientCall import io.grpc.Metadata import io.grpc.Metadata.Key import io.grpc.MethodDescriptor import io.grpc.ClientInterceptorsobject POSClient {class AuthClientInterceptor(jwt: String) extends ClientInterceptor {def interceptCall[ReqT, RespT](methodDescriptor: MethodDescriptor[ReqT, RespT], callOptions: CallOptions, channel: io.grpc.Channel): ClientCall[ReqT, RespT] =new ForwardingClientCall.SimpleForwardingClientCall[ReqT, RespT](channel.newCall(methodDescriptor, callOptions)) {override def start(responseListener: ClientCall.Listener[RespT], headers: Metadata): Unit = {headers.put(Key.of("jwt", Metadata.ASCII_STRING_MARSHALLER), jwt)super.start(responseListener, headers)}}}def main(args: Array[String]): Unit = {//build connection channelval unsafeChannel = NettyChannelBuilder.forAddress("192.168.0.189",50051).negotiationType(NegotiationType.PLAINTEXT).build()val authClient = SendCommandGrpc.blockingStub(unsafeChannel)val jwt = authClient.getAuthToken(PBPOSCredential(userid="johnny",password="p4ssw0rd")).jwtprintln(s"got jwt: $jwt")val securedChannel = ClientInterceptors.intercept(unsafeChannel, new AuthClientInterceptor(jwt))val securedClient = SendCommandGrpc.blockingStub(securedChannel)val resp = securedClient.singleResponse(PBPOSCommand())println(s"secured response: $resp")// wait for async executionscala.io.StdIn.readLine()}}

轉(zhuǎn)載于:https://www.cnblogs.com/tiger-xc/p/11188900.html

總結(jié)

以上是生活随笔為你收集整理的Akka-CQRS（16）- gRPC用JWT进行权限管理的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： Zookeeper-watcher机制源
下一篇： linux dd安装系统,通过DD命令安