How to set & retrieve a user ID after authenicating?

[bean...@googlemail.com]

I've created client & server interceptors. The Client interceptor injects a mock auth token which the server verifies. Here's the Server Interceptor:

class ServerAuthInterceptor extends ServerInterceptor {
  private val AUTHORIZATION: Metadata.Key[String] = Metadata.Key.of("Authorization",
    Metadata.ASCII_STRING_MARSHALLER)

  override def interceptCall[RequestT, ResponseT](
       methodName: String, serverCall: ServerCall[ResponseT], headers: Headers,
       serverCallHandler: ServerCallHandler[RequestT, ResponseT]): Listener[RequestT] = {

    val authHeader = headers.get(AUTHORIZATION)

    // todo: Hook this up to a real auth provider
    assert(authHeader == "Token: abc")    // == User 1

    serverCallHandler.startCall(methodName, serverCall, headers)
  }
}

How can I inject the User Id into the call and access it later? I guess I could just add another header, but how could I retrieve that later from within an RPC method? Ordinarily I could use a singleton, but with all the concurrency going on that would be a mistake.

Thanks

[Eric Anderson]

Our recommendation is to use ThreadLocal that is set/cleared on entry/exit of onPayload. You can hide the ThreadLocal by injecting a Provider.

See https://github.com/grpc/grpc-java/issues/174#issuecomment-77938347 (and the whole thread) for a few details on rationale.

--
You received this message because you are subscribed to the Google Groups "grpc.io" group.
To unsubscribe from this group and stop receiving emails from it, send an email to grpc-io+u...@googlegroups.com.
To post to this group, send email to grp...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/grpc-io/bab04e06-86c5-4176-aea0-cfacea14d466%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[bean...@googlemail.com]

OK makes sense.

I quite work out how to override onPayload to set/clear the threadlocal. I've been trying to return a class that extends ServerInterceptors.ForwardingListener but I can't work out how to get a `ServerCall.Listener<RespT>` to delegate to. I can only work out how to get a ServerCall.Listener<RequestT>.

Please can you give me some pointers?

Thanks

[Eric Anderson]

On Wed, Apr 1, 2015 at 12:16 PM, <bean...@googlemail.com> wrote:

I quite work out how to override onPayload to set/clear the threadlocal. I've been trying to return a class that extends ServerInterceptors.ForwardingListener but I can't work out how to get a `ServerCall.Listener<RespT>` to delegate to. I can only work out how to get a ServerCall.Listener<RequestT>.

Please can you give me some pointers?

You should only need a Listener<RequestT>.

public <RequestT, ResponseT> ServerCall.Listener<RequestT> interceptCall(

      String method, ServerCall<ResponseT> call,

      Metadata.Headers headers,

      ServerCallHandler<RequestT, ResponseT> next) {

  return new MyThreadLocalApplyingListener<RequestT>(

    next.startCall(method, call, headers));

}

[bean...@googlemail.com]

Sorry to be a pain, but I'm struggling with this. Here's my interceptor:

package com.example.api.server

import javax.inject.Provider

import io.grpc.Metadata.Headers
import io.grpc.ServerCall.Listener
import io.grpc.ServerInterceptors.ForwardingListener
import io.grpc._

/**
 * Validate auth credentials.
 */

class ServerAuthInterceptor extends ServerInterceptor {

  private val AUTHORIZATION: Metadata.Key[String] = Metadata.Key.of("Authorization",
    Metadata.ASCII_STRING_MARSHALLER)

  private val threadLocalUserId: ThreadLocal[Int] = new ThreadLocal[Int]()

  override def interceptCall[RequestT, ResponseT](

       method: String, call: ServerCall[ResponseT], headers: Headers,
       next: ServerCallHandler[RequestT, ResponseT]): Listener[RequestT] = {

    val authHeader = headers.get(AUTHORIZATION)

    // todo: Hook this up to a real auth provider

    assert(authHeader == "UserId: 1")

    val userId = 1

    new ForwardingListener[RequestT](next.startCall(method, call, headers)) {
      override def onPayload(payload: RequestT): Unit = {

        // set the threadlocal before the call
        threadLocalUserId.set(userId)

        super.onPayload(payload)

        // now clear it
        threadLocalUserId.remove()
      }
    }
  }

  /**
   * @return A Provider to allow access to the User ID.
   */
  def userIdProvider(): Provider[Int] = {

    new Provider[Int] {

      override def get(): Int = {
        threadLocalUserId.get()
      }
    }
  }
}

And here's how I create the server:

val authInterceptor = new ServerAuthInterceptor()

val serviceDefinition = MyGrpc.bindService(
  new MyApiService(dataStore = new DbDataStore(),
    userIdProvider = authInterceptor.userIdProvider()))

val gRpcServerWithAuth = ServerInterceptors.intercept(serviceDefinition,
  authInterceptor)

gRpcServer = NettyServerBuilder.forPort(port)
  .addService(gRpcServerWithAuth)
  .build()

The problem is that in my RPC methods, the userIdProvider returns 0 instead of 1. I've set a breakpoint on the call to threadLocalUserId.set(userId) and can see it's being set, but it seems to be being unset again before my gRPC method is actually called. What am I doing wrong?

[Eric Anderson]

On Thu, Apr 2, 2015 at 5:25 AM, <bean...@googlemail.com> wrote:

Sorry to be a pain, but I'm struggling with this.

I'm glad you asked more. The reason why it isn't working isn't obvious. I had to code this up and was equally surprised it didn't work. After digging in, I found out why.

    new ForwardingListener[RequestT](next.startCall(method, call, headers)) {
      override def onPayload(payload: RequestT): Unit = {

The problem is we don't call the service implementation until onHalfClose(), not on onPayload(). You can see a comment there saying we do so on purpose.

So to fix your ForwardingListener, also override onHalfClose and set the ThreadLocal there as well. With that change my interceptor started functioning as expected.

Interestingly, the comment in ServerCalls saying why we don't deliver immediately is no longer true; the restriction it is referring to was lifted before open sourcing. We may still choose to wait until onHalfClose() for other reasons, but we'll need to at least change the comment. I've created issue 263 to fix that.

Sorry!

[bean...@googlemail.com]

Thanks it's working now.

[bean...@googlemail.com]

I'm having problems with this approach using an async stub. The interceptor on the blocking stub works correctly, and the user ID is correctly forwarded to the server. However, when I use the same channel to instantiate an async stub and then make a request, the user ID isn't being correctly forwarded to the server. I guess that a closure that should be being created isn't being. Here's my code:

val interceptor = new ClientAuthInterceptor(userId)
val channelWithAuth = ClientInterceptors.intercept(channel, interceptor)

blockingStub = SponventGrpc.newBlockingStub(channelWithAuth)
asyncStub = SponventGrpc.newStub(channelWithAuth)         // doesn't forward the user ID - the default value of 0 is sent to the server when this stub is used

Is there a bug in the Stub implementation somewhere, or am I not creating asyncStub correctly?

Thanks

[bean...@googlemail.com]

Hmm... Actually there's something weird going on. I've got two methods that i'm using the async stub for, and for one of them the user ID is correctly forwarded, but for the other it always defaults to 0. I'll recheck my code, but AFAICT I'm using the same code in both places so it's not immediately clear where the error lies...

[bean...@googlemail.com]

OK so the problem is on the server side. I have a method that accepts a stream from the client. It has an RPC definition of:

rpc sendMessage(stream Message) returns (Empty) {}

It seems that for methods that accept streams like this, onPayload and onHalfClose aren't called, so the value of the user ID isn't being set on the threadLocal. So the following code in my interceptor never gets called for this particular method. Since this is the only method I have that accepts a stream from the client, I guess it's treated differently somehow. 

What do I need to update in the following so that the user ID will be set on the thread local for these sorts of methods?

new ForwardingListener[RequestT](next.startCall(method, call, headers)) {
  override def onPayload(payload: RequestT): Unit = {

    // set the threadlocal before the call
    threadLocalUserId.set(userId)

    super.onPayload(payload)

    // now clear it
    threadLocalUserId.remove()
  }

  override def onHalfClose(): Unit = {

    // set the threadlocal before the call
    threadLocalUserId.set(userId)

    super.onHalfClose()

    // now clear it
    threadLocalUserId.remove()
  }
}

Thanks

[Eric Anderson]

Another great question.

You probably need to set threadLocalUserId within startCall(), also. For unary requests we don't call your service within startCall, but for streaming we do.

--

You received this message because you are subscribed to the Google Groups "grpc.io" group.
To unsubscribe from this group and stop receiving emails from it, send an email to grpc-io+u...@googlegroups.com.
To post to this group, send email to grp...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/grpc-io/d8d02a1f-c20b-43df-8de5-e9c101b54ea0%40googlegroups.com.

[bean...@googlemail.com]

Thanks Eric, I'll try decorating my call to `next.startCall` to set the threadLocal first.