Tiny GCP library for the service account to get the identity token

gcp/service-auth/.gitignore

@@ -0,0 +1,2 @@
+browse-local.json
+test.log

gcp/service-auth/README.md

@@ -0,0 +1,11 @@
+# GCP Service Account Auth Token Tiny Library
+
+A library to acquire an identity auth token of service account for authenticated services.
+Expected to be used for GCP functions, Cloud runs and also at CIT web node.
+
+## Service account role requirements
+
+When using this, the library gets the id toke of the service account. For non-GCP, 
+you can use GOOGLE_APPLICATION_CREDENTIALS to point to the SA credentials.
+
+The service account needs "Cloud Run Invoker" role

gcp/service-auth/gcp_service_auth/__init__.py

@@ -0,0 +1 @@
+from .service_auth import *

gcp/service-auth/gcp_service_auth/service_auth.py

@@ -0,0 +1,47 @@
+"""
+Acquire gcp service account auth token.
+This is expected to use the default SA cred.
+
+Make sure that the service account has the cloud run invoker role enabled if used to talk to
+the service that needs auth.
+"""
+import typing
+import datetime
+
+import logging
+from google.auth.credentials import Credentials as GcpCredentials
+import google.auth.transport.requests
+import google.oauth2.id_token
+
+class GcpIdentityToken:
+    _credentials: GcpCredentials
+    _project: typing.Any
+    _last_refresh: datetime.datetime
+    target: str
+    _token: str
+    _logger: logging.Logger | None
+    expiration: datetime.timedelta
+
+    def __init__(self, target: str, logger: logging.Logger | None =None,
+                 expiration: datetime.timedelta = datetime.timedelta(minutes=30)):
+        self.expiration = expiration
+        self.target = target
+        self._logger = logger
+        self._credentials, self._project = google.auth.default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
+        self.refresh()
+        pass
+
+    def refresh(self) -> None:
+        """Refresh the token"""
+        self._last_refresh = datetime.datetime.utcnow()
+        auth_req = google.auth.transport.requests.Request()
+        self._token = google.oauth2.id_token.fetch_id_token(auth_req, self.target)
+        if self._logger:
+            self._logger.info("Token refreshed")
+        pass
+
+    @property
+    def token(self) -> str:
+        if datetime.datetime.utcnow() - self._last_refresh > self.expiration:
+            self.refresh()
+        return self._token

gcp/service-auth/pyproject.toml

@@ -0,0 +1,25 @@
+[tool.poetry]
+name = "gcp-service-auth"
+version = "0.1.0"
+description = "Inspecting TeX in arXiv submissions"
+authors = ["arxiv.org"]
+license = "BSD-3 License"
+readme = "README.md"
+packages = [
+    { include = "gcp_service_auth" }
+]
+include = ["README.md"]
+
+[tool.poetry.dependencies]
+python = "^3.8"
+google-auth = "^2.29.0"
+
+[tool.poetry.group.dev.dependencies]
+pytest = "^8.1.1"
+mypy = "*"
+mypy-extensions = "*"
+pydantic = "1.10.*"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

gcp/service-auth/pytest.ini

@@ -0,0 +1,4 @@
+[pytest]
+markers =
+    with_op: marks tests to run with 1password CLI
+

gcp/service-auth/tests/test_get_token.py

@@ -0,0 +1,39 @@
+import os
+import subprocess
+import time
+from typing import Generator
+
+import pytest
+import logging
+import datetime
+from pathlib import Path
+
+from gcp_service_auth import GcpIdentityToken
+
+dev_target = "https://gcp-genpdf-6lhtms3oua-uc.a.run.app"
+
+@pytest.fixture(scope="module")
+#@pytest.mark.with_op
+def gcp_browse_cred() -> Generator[str, str, str]:
+    logging.basicConfig(level=logging.DEBUG)
+    cred_file = os.path.join(Path(__file__).parent, "browse-local.json")
+    cred_file = cred_file
+    if not os.path.exists(cred_file):
+        subprocess.run(["op", "document", "get", "4feibaz4tzn6iwk5c7ggvb7xwi", "-o", cred_file])
+    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = cred_file
+    yield cred_file
+    os.remove(cred_file)
+    return ""
+
+@pytest.mark.with_op
+def test_get_token(gcp_browse_cred: str) -> None:
+    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = gcp_browse_cred
+    logger = logging.getLogger("test")
+    idt = GcpIdentityToken(dev_target, logger=logger,
+                           expiration=datetime.timedelta(seconds=1))
+    token0 = idt.token
+    time.sleep(2)
+    token1 = idt.token
+    assert token0 is not None
+    assert token1 is not None
+    assert token0 != token1