This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new f37586c84 huawei-push-kit: use expires_in from responses (#1812)
f37586c84 is described below
commit f37586c84b31b26b43de16e4299b9c0197661975
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Aug 10 10:07:35 2026 +0100
huawei-push-kit: use expires_in from responses (#1812)
---
.../google/auth/GoogleComputeMetadata.scala | 13 ++++++--
.../connectors/google/auth/GoogleOAuth2.scala | 16 +++++++--
.../google/auth/UserAccessMetadata.scala | 13 ++++++--
.../huawei/pushkit/impl/HmsTokenApi.scala | 11 ++++--
.../huawei/pushkit/impl/HmsTokenApiSpec.scala | 39 ++++++++++++++++++++++
5 files changed, 83 insertions(+), 9 deletions(-)
diff --git
a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala
b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala
index 6336b74aa..dffb07ead 100644
---
a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala
+++
b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleComputeMetadata.scala
@@ -23,6 +23,7 @@ import pekko.http.scaladsl.model.{ HttpRequest, Uri }
import pekko.http.scaladsl.model.headers.RawHeader
import pekko.http.scaladsl.unmarshalling.Unmarshal
import pekko.stream.Materializer
+import pdi.jwt.JwtTime
import java.time.Clock
import scala.concurrent.Future
@@ -55,8 +56,16 @@ private[auth] object GoogleComputeMetadata {
implicit val system: ActorSystem = mat.system
for {
response <- Http().singleRequest(tokenRequest(scopes))
- token <- Unmarshal(response.entity).to[AccessToken]
- } yield token
+ tokenResponse <- Unmarshal(response.entity).to[AccessTokenResponse]
+ } yield {
+ val expiresIn = if (tokenResponse.expires_in > 0)
tokenResponse.expires_in
+ else {
+ system.log.warning("Google OAuth2 response contained invalid
expires_in ({}), falling back to default {}s",
+ tokenResponse.expires_in, GoogleOAuth2.DefaultExpiresIn)
+ GoogleOAuth2.DefaultExpiresIn
+ }
+ AccessToken(tokenResponse.access_token, JwtTime.nowSeconds + expiresIn)
+ }
}
def getProjectId()(
diff --git
a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
index 6eab497ba..a879f9aa8 100644
---
a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
+++
b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/GoogleOAuth2.scala
@@ -23,19 +23,20 @@ import pekko.stream.Materializer
import pekko.stream.connectors.google.http.GoogleHttp
import pekko.stream.connectors.google.jwt.JwtSprayJson
import pekko.stream.connectors.google.{ implicits, RequestSettings }
+import pdi.jwt.{ JwtClaim, JwtTime }
import pdi.jwt.JwtAlgorithm.RS256
-import pdi.jwt.JwtClaim
import spray.json.DefaultJsonProtocol._
import spray.json.JsonFormat
import java.time.Clock
-import scala.concurrent.Future
+import scala.concurrent.{ ExecutionContext, Future }
import scala.util.control.NonFatal
@InternalApi
private[auth] object GoogleOAuth2 {
private val oAuthTokenUrl = "https://oauth2.googleapis.com/token";
+ private[auth] val DefaultExpiresIn = 3600
def getAccessToken(clientEmail: String, privateKey: String, scopes:
Set[String])(
implicit mat: Materializer,
@@ -51,7 +52,16 @@ private[auth] object GoogleOAuth2 {
"grant_type" -> "urn:ietf:params:oauth:grant-type:jwt-bearer",
"assertion" -> generateJwt(clientEmail, privateKey, scopes)).toEntity
- GoogleHttp().singleRequest[AccessToken](HttpRequest(POST, oAuthTokenUrl,
entity = entity))
+ GoogleHttp().singleRequest[AccessTokenResponse](HttpRequest(POST,
oAuthTokenUrl, entity = entity)).map {
+ case AccessTokenResponse(access_token, _, expires_in) =>
+ val safeExpiresIn = if (expires_in > 0) expires_in
+ else {
+ system.log.warning("Google OAuth2 response contained invalid
expires_in ({}), falling back to default {}s",
+ expires_in, DefaultExpiresIn)
+ DefaultExpiresIn
+ }
+ AccessToken(access_token, JwtTime.nowSeconds + safeExpiresIn)
+ }(ExecutionContext.parasitic)
} catch {
case NonFatal(e) =>
Future.failed(e)
diff --git
a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessMetadata.scala
b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessMetadata.scala
index c3efbfcce..193da9f7c 100644
---
a/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessMetadata.scala
+++
b/google-common/src/main/scala/org/apache/pekko/stream/connectors/google/auth/UserAccessMetadata.scala
@@ -23,6 +23,7 @@ import pekko.http.scaladsl.model.headers.RawHeader
import pekko.http.scaladsl.model.{ FormData, HttpRequest }
import pekko.http.scaladsl.unmarshalling.Unmarshal
import pekko.stream.Materializer
+import pdi.jwt.JwtTime
import java.time.Clock
import scala.concurrent.Future
@@ -49,7 +50,15 @@ private[auth] object UserAccessMetadata {
implicit val system: ActorSystem = mat.system
for {
response <- Http().singleRequest(tokenRequest(clientId, clientSecret,
refreshToken))
- token <- Unmarshal(response.entity).to[AccessToken]
- } yield token
+ tokenResponse <- Unmarshal(response.entity).to[AccessTokenResponse]
+ } yield {
+ val expiresIn = if (tokenResponse.expires_in > 0)
tokenResponse.expires_in
+ else {
+ system.log.warning("Google OAuth2 response contained invalid
expires_in ({}), falling back to default {}s",
+ tokenResponse.expires_in, GoogleOAuth2.DefaultExpiresIn)
+ GoogleOAuth2.DefaultExpiresIn
+ }
+ AccessToken(tokenResponse.access_token, JwtTime.nowSeconds + expiresIn)
+ }
}
}
diff --git
a/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApi.scala
b/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApi.scala
index 509843a7d..170f818b8 100644
---
a/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApi.scala
+++
b/huawei-push-kit/src/main/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApi.scala
@@ -36,6 +36,7 @@ import scala.concurrent.Future
private[pushkit] class HmsTokenApi(http: => HttpExt, system: ActorSystem,
forwardProxy: Option[ForwardProxy]) {
import PushKitJsonSupport._
+ private val log = system.log
private val authUrl = "https://oauth-login.cloud.huawei.com/oauth2/v3/token";
def now: Long = JwtTime.nowSeconds(Clock.systemUTC())
@@ -43,7 +44,6 @@ private[pushkit] class HmsTokenApi(http: => HttpExt, system:
ActorSystem, forwar
def getAccessToken(clientId: String, privateKey: String)(
implicit materializer: Materializer): Future[AccessTokenExpiry] = {
import materializer.executionContext
- val expiresAt = now + 3600
val requestEntity = FormData(
"grant_type" -> "client_credentials",
@@ -60,9 +60,15 @@ private[pushkit] class HmsTokenApi(http: => HttpExt, system:
ActorSystem, forwar
}
result <- Unmarshal(response.entity).to[OAuthResponse]
} yield {
+ val expiresIn = if (result.expires_in > 0) result.expires_in
+ else {
+ log.warning("Huawei OAuth2 response contained invalid expires_in ({}),
falling back to default {}s",
+ result.expires_in, HmsTokenApi.DefaultExpiresIn)
+ HmsTokenApi.DefaultExpiresIn
+ }
AccessTokenExpiry(
accessToken = result.access_token,
- expiresAt = expiresAt)
+ expiresAt = now + expiresIn)
}
}
}
@@ -72,6 +78,7 @@ private[pushkit] class HmsTokenApi(http: => HttpExt, system:
ActorSystem, forwar
*/
@InternalApi
private[pushkit] object HmsTokenApi {
+ val DefaultExpiresIn = 3600
case class AccessTokenExpiry(accessToken: String, expiresAt: Long) {
override def toString: String =
"AccessTokenExpiry(accessToken=*****," +
diff --git
a/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApiSpec.scala
b/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApiSpec.scala
index 99f46463b..c368b3f7b 100644
---
a/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApiSpec.scala
+++
b/huawei-push-kit/src/test/scala/org/apache/pekko/stream/connectors/huawei/pushkit/impl/HmsTokenApiSpec.scala
@@ -103,6 +103,45 @@ class HmsTokenApiSpec
case AccessTokenExpiry("token", exp) if exp >
(System.currentTimeMillis / 1000L + 3000L) =>
}
}
+
+ "use expires_in from server response" in {
+ val http = mock[HttpExt]
+ when(
+ http.singleRequest(any[HttpRequest](),
+ any[HttpsConnectionContext](),
+ any[ConnectionPoolSettings](),
+ any[LoggingAdapter]())).thenReturn(
+ Future.successful(
+ HttpResponse(
+ entity = HttpEntity(ContentTypes.`application/json`,
+ """{"access_token": "token", "token_type": "String",
"expires_in": 7200}"""))))
+
+ val api = new HmsTokenApi(http, system, Option.empty)
+ val result = api.getAccessToken(config.appId,
config.appSecret).futureValue
+ result.accessToken shouldBe "token"
+ // expires_in=7200 means expiry should be ~7200s from now, well beyond
the old hardcoded 3600s
+ result.expiresAt should be > (System.currentTimeMillis / 1000L + 6000L)
+ }
+
+ "fall back to default expiry when expires_in is invalid" in {
+ val http = mock[HttpExt]
+ when(
+ http.singleRequest(any[HttpRequest](),
+ any[HttpsConnectionContext](),
+ any[ConnectionPoolSettings](),
+ any[LoggingAdapter]())).thenReturn(
+ Future.successful(
+ HttpResponse(
+ entity = HttpEntity(ContentTypes.`application/json`,
+ """{"access_token": "token", "token_type": "String",
"expires_in": 0}"""))))
+
+ val api = new HmsTokenApi(http, system, Option.empty)
+ val result = api.getAccessToken(config.appId,
config.appSecret).futureValue
+ result.accessToken shouldBe "token"
+ // expires_in=0 should fall back to default 3600s
+ result.expiresAt should be > (System.currentTimeMillis / 1000L + 3000L)
+ result.expiresAt should be < (System.currentTimeMillis / 1000L + 4000L)
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
