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

   前面谈过gRPC的SSL/TLS安全机制，发现设置过程比较复杂：比如证书签名：需要服务端、客户端两头都设置等。想想实际上用JWT会更加便捷，而且更安全和功能强大，因为除JWT的加密签名之外还可以把私密的用户信息放在JWT里加密后在服务端和客户端之间传递。当然，最基本的是通过对JWT的验证机制可以控制客户端对某些功能的使用权限。

通过JWT实现gRPC的函数调用权限管理原理其实很简单：客户端首先从服务端通过身份验证获取JWT，然后在调用服务函数时把这个JWT同时传给服务端进行权限验证。客户端提交身份验证请求返回JWT可以用一个独立的服务函数实现，如下面.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从客户端传送至服务端，因为gRPC基本上骑劫了Request和Response。其中一个方法是通过Interceptor来截取Request的header即metadata。客户端将JWT写入metadata，服务端从metadata读取JWT。

我们先看看客户端的Interceptor设置和使用：

  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())

身份验证请求即JWT获取是不需要Interceptor的，所以要用没有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的构建和使用已经在前面的几篇博文里讨论过了： 

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)}}}

服务端Interceptor的构建和设置如下： 

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只能在构建server时接触到，在具体服务函数里是无法调用request-header的，但gRPC又一个结构Context可以在两个地方都能调用。所以，我们可以在构建server时把JWT从header搬到Context里。不过，千万注意这个Context的读写必须在同一个线程里。在服务端的Interceptor里我们把JWT从metadata里读出然后写入Context。在需要权限管理的服务函数里再从Context里读取JWT进行验证： 

   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的构建也是一个服务函数： 

   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!"))}}

还需要一个模拟的身份验证服务函数： 

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 {      //单据状态string opr  = 1;    //收款员int64  jseq = 2;    //begin journal sequence for read-side replayint32  num  = 3;    //当前单号int32  seq  = 4;    //当前序号bool   void = 5;    //取消模式bool   refd = 6;    //退款模式bool   susp = 7;    //挂单bool   canc = 8;    //废单bool   due  = 9;    //当前余额string su   = 10;   //主管编号string mbr  = 11;   //会员号int32  mode = 12;   //当前操作流程：0=logOff, 1=LogOn, 2=Payment
}message PBTxnItem {       //交易记录string txndate    = 1;   //交易日期string txntime    = 2;   //录入时间string opr        = 3;   //操作员int32  num        = 4;   //销售单号int32  seq        = 5;   //交易序号int32  txntype    = 6;   //交易类型int32  salestype  = 7;   //销售类型int32  qty        = 8;   //交易数量int32  price      = 9;   //单价（分）int32  amount     = 10;  //码洋（分）int32  disc       = 11;  //折扣率 (%)int32  dscamt     = 12;  //折扣额：负值  net实洋 = amount + dscamtstring member     = 13;  //会员卡号string code       = 14;  //编号（商品、卡号...）string acct       = 15;  //账号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

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)}}}

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()}}