Authentication for push subscriptions

/// <summary>
/// Extended JWT payload to match the pubsub payload format.
/// </summary>
publicclassPubSubPayload:JsonWebSignature.Payload
{
[JsonProperty("email")]
publicstringEmail{get;set;}
[JsonProperty("email_verified")]
publicstringEmailVerified{get;set;}
}
/// <summary>
/// Handle authenticated push request coming from pubsub.
/// See the full sample in https://github.com/GoogleCloudPlatform/dotnet-docs-samples/blob/main/appengine/flexible/Pubsub/Pubsub.Sample/Controllers/HomeController.cs
/// </summary>
[HttpPost]
[Route("/AuthPush")]
publicasyncTask<IActionResult>AuthPushAsync([FromBody]PushBodybody,[FromQuery]stringtoken)
{
// Get the Cloud Pub/Sub-generated "Authorization" header.
stringauthorizaionHeader=HttpContext.Request.Headers["Authorization"];
stringverificationToken=token??body.message.attributes["token"];
// JWT token comes in `Bearer <JWT>` format substring 7 specifies the position of first JWT char.
stringauthToken=authorizaionHeader.StartsWith("Bearer ")?authorizaionHeader.Substring(7):null;
if(verificationToken!=_options.VerificationToken||authTokenisnull)
{
returnnewBadRequestResult();
}
// Verify and decode the JWT.
// Note: For high volume push requests, it would save some network
// overhead if you verify the tokens offline by decoding them using
// Google's Public Cert; caching already seen tokens works best when
// a large volume of messages have prompted a single push server to
// handle them, in which case they would all share the same token for
// a limited time window.
varpayload=awaitJsonWebSignature.VerifySignedTokenAsync<PubSubPayload>(authToken);
// IMPORTANT: you should validate payload details not covered
// by signature and audience verification above, including:
// - Ensure that `payload.Email` is equal to the expected service
// account set up in the push subscription settings.
// - Ensure that `payload.Email_verified` is set to true.
varmessageBytes=Convert.FromBase64String(body.message.data);
stringmessage=System.Text.Encoding.UTF8.GetString(messageBytes);
s_authenticatedMessages.Add(message);
returnnewOkResult();
}

// receiveMessagesHandler validates authentication token and caches the Pub/Sub
// message received.
func(a*app)receiveMessagesHandler(whttp.ResponseWriter,r*http.Request){
ifr.Method!="POST"{
http.Error(w,http.StatusText(http.StatusMethodNotAllowed),http.StatusMethodNotAllowed)
return
}
// Verify that the request originates from the application.
// a.pubsubVerificationToken = os.Getenv("PUBSUB_VERIFICATION_TOKEN")
iftoken,ok:=r.URL.Query()["token"];!ok||len(token)!=1||token[0]!=a.pubsubVerificationToken{
http.Error(w,"Bad token",http.StatusBadRequest)
return
}
// Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
authHeader:=r.Header.Get("Authorization")
ifauthHeader==""||len(strings.Split(authHeader," "))!=2{
http.Error(w,"Missing Authorization header",http.StatusBadRequest)
return
}
token:=strings.Split(authHeader," ")[1]
// Verify and decode the JWT.
// If you don't need to control the HTTP client used you can use the
// convenience method idtoken.Validate instead of creating a Validator.
v,err:=idtoken.NewValidator(r.Context(),option.WithHTTPClient(a.defaultHTTPClient))
iferr!=nil{
http.Error(w,"Unable to create Validator",http.StatusBadRequest)
return
}
// Please change http://example.com to match with the value you are
// providing while creating the subscription.
payload,err:=v.Validate(r.Context(),token,"http://example.com")
iferr!=nil{
http.Error(w,fmt.Sprintf("Invalid Token: %v",err),http.StatusBadRequest)
return
}
ifpayload.Issuer!="accounts.google.com" && payload.Issuer!="https://accounts.google.com"{
http.Error(w,"Wrong Issuer",http.StatusBadRequest)
return
}
// IMPORTANT: you should validate claim details not covered by signature
// and audience verification above, including:
// - Ensure that `payload.Claims["email"]` is equal to the expected service
// account set up in the push subscription settings.
// - Ensure that `payload.Claims["email_verified"]` is set to true.
ifpayload.Claims["email"]!="test-service-account-email@example.com"||payload.Claims["email_verified"]!=true{
http.Error(w,"Unexpected email identity",http.StatusBadRequest)
return
}
varprpushRequest
iferr:=json.NewDecoder(r.Body).Decode(&pr);err!=nil{
http.Error(w,fmt.Sprintf("Could not decode body: %v",err),http.StatusBadRequest)
return
}
a.messagesMu.Lock()
defera.messagesMu.Unlock()
// Limit to ten.
a.messages=append(a.messages,pr.Message.Data)
iflen(a.messages) > maxMessages{
a.messages=a.messages[len(a.messages)-maxMessages:]
}
fmt.Fprint(w,"OK")
}

@WebServlet(value="/pubsub/authenticated-push")
publicclass PubSubAuthenticatedPushextendsHttpServlet{
privatefinalStringpubsubVerificationToken=System.getenv("PUBSUB_VERIFICATION_TOKEN");
privatefinalMessageRepositorymessageRepository;
privatefinalGoogleIdTokenVerifierverifier=
newGoogleIdTokenVerifier.Builder(newNetHttpTransport(),newGsonFactory())
/**
 * Please change example.com to match with value you are providing while creating
 * subscription as provided in @see <a
 * href="https://github.com/GoogleCloudPlatform/java-docs-samples/tree/main/appengine-java8/pubsub">README</a>.
 */
.setAudience(Collections.singletonList("example.com"))
.build();
privatefinalGsongson=newGson();
@Override
publicvoiddoPost(HttpServletRequestreq,HttpServletResponseresp)
throwsIOException,ServletException{
// Verify that the request originates from the application.
if(req.getParameter("token").compareTo(pubsubVerificationToken)!=0){
resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
return;
}
// Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
StringauthorizationHeader=req.getHeader("Authorization");
if(authorizationHeader==null
||authorizationHeader.isEmpty()
||authorizationHeader.split(" ").length!=2){
resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
return;
}
Stringauthorization=authorizationHeader.split(" ")[1];
try{
// Verify and decode the JWT.
// Note: For high volume push requests, it would save some network overhead
// if you verify the tokens offline by decoding them using Google's Public
// Cert; caching already seen tokens works best when a large volume of
// messsages have prompted a single push server to handle them, in which
// case they would all share the same token for a limited time window.
GoogleIdTokenidToken=verifier.verify(authorization);
GoogleIdToken.Payloadpayload=idToken.getPayload();
// IMPORTANT: you should validate claim details not covered by signature
// and audience verification above, including:
// - Ensure that `payload.getEmail()` is equal to the expected service
// account set up in the push subscription settings.
// - Ensure that `payload.getEmailVerified()` is set to true.
messageRepository.saveToken(authorization);
messageRepository.saveClaim(payload.toPrettyString());
// parse message object from "message" field in the request body json
// decode message data from base64
Messagemessage=getMessage(req);
messageRepository.save(message);
// 200, 201, 204, 102 status codes are interpreted as success by the Pub/Sub system
resp.setStatus(102);
super.doPost(req,resp);
}catch(Exceptione){
resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
}
}
privateMessagegetMessage(HttpServletRequestrequest)throwsIOException{
StringrequestBody=request.getReader().lines().collect(Collectors.joining("\n"));
JsonElementjsonRoot=JsonParser.parseString(requestBody).getAsJsonObject();
StringmessageStr=jsonRoot.getAsJsonObject().get("message").toString();
Messagemessage=gson.fromJson(messageStr,Message.class);
// decode from base64
Stringdecoded=decode(message.getData());
message.setData(decoded);
returnmessage;
}
privateStringdecode(Stringdata){
returnnewString(Base64.getDecoder().decode(data));
}
PubSubAuthenticatedPush(MessageRepositorymessageRepository){
this.messageRepository=messageRepository;
}
publicPubSubAuthenticatedPush(){
this(MessageRepositoryImpl.getInstance());
}
}

app.post('/pubsub/authenticated-push',jsonBodyParser,async(req,res)=>{
// Verify that the request originates from the application.
if(req.query.token!==PUBSUB_VERIFICATION_TOKEN){
res.status(400).send('Invalid request');
return;
}
// Verify that the push request originates from Cloud Pub/Sub.
try{
// Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
constbearer=req.header('Authorization');
const[,token]=bearer.match(/Bearer (.*)/);
tokens.push(token);
// Verify and decode the JWT.
// Note: For high volume push requests, it would save some network
// overhead if you verify the tokens offline by decoding them using
// Google's Public Cert; caching already seen tokens works best when
// a large volume of messages have prompted a single push server to
// handle them, in which case they would all share the same token for
// a limited time window.
constticket=awaitauthClient.verifyIdToken({
idToken:token,
audience:'example.com',
});
constclaim=ticket.getPayload();
// IMPORTANT: you should validate claim details not covered
// by signature and audience verification above, including:
// - Ensure that `claim.email` is equal to the expected service
// account set up in the push subscription settings.
// - Ensure that `claim.email_verified` is set to true.
claims.push(claim);
}catch(e){
res.status(400).send('Invalid token');
return;
}
// The message is a unicode string encoded in base64.
constmessage=Buffer.from(req.body.message.data,'base64').toString(
'utf-8'
);
messages.push(message);
res.status(200).send();
});

@app.route("/push-handlers/receive_messages", methods=["POST"])
defreceive_messages_handler():
 # Verify that the request originates from the application.
 if request.args.get("token", "") != current_app.config["PUBSUB_VERIFICATION_TOKEN"]:
 return "Invalid request", 400
 # Verify that the push request originates from Cloud Pub/Sub.
 try:
 # Get the Cloud Pub/Sub-generated JWT in the "Authorization" header.
 bearer_token = request.headers.get("Authorization")
 token = bearer_token.split(" ")[1]
 TOKENS.append(token)
 # Verify and decode the JWT. `verify_oauth2_token` verifies
 # the JWT signature, the `aud` claim, and the `exp` claim.
 # Note: For high volume push requests, it would save some network
 # overhead if you verify the tokens offline by downloading Google's
 # Public Cert and decode them using the `google.auth.jwt` module;
 # caching already seen tokens works best when a large volume of
 # messages have prompted a single push server to handle them, in which
 # case they would all share the same token for a limited time window.
 claim = id_token.verify_oauth2_token(
 token, requests.Request(), audience="example.com"
 )
 # IMPORTANT: you should validate claim details not covered by signature
 # and audience verification above, including:
 # - Ensure that `claim["email"]` is equal to the expected service
 # account set up in the push subscription settings.
 # - Ensure that `claim["email_verified"]` is set to true.
 CLAIMS.append(claim)
 except Exception as e:
 return f"Invalid token: {e}\n", 400
 envelope = json.loads(request.data.decode("utf-8"))
 payload = base64.b64decode(envelope["message"]["data"])
 MESSAGES.append(payload)
 # Returning any 2xx status indicates successful receipt of the message.
 return "OK", 200

post"/pubsub/authenticated-push"do
halt400ifparams[:token]!=PUBSUB_VERIFICATION_TOKEN
begin
bearer=request.env["HTTP_AUTHORIZATION"]
token= /Bearer (.*)/.match(bearer)[1]
claim=Google::Auth::IDTokens .verify_oidc token,aud:"example.com"
# IMPORTANT: you should validate claim details not covered by signature
# and audience verification above, including:
# - Ensure that `claim["email"]` is equal to the expected service
# account set up in the push subscription settings.
# - Ensure that `claim["email_verified"]` is set to true.
claims.pushclaim
rescueGoogle::Auth::IDTokens ::VerificationError =>e
puts"VerificationError: #{e.message}"
halt400,"Invalid token"
end
message=JSON.parserequest.body.read
payload=Base64.decode64message["message"]["data"]
messages.pushpayload
end