Merge branch 'add_bypass' into 'main'

Update 2 files

See merge request !18

dinamis_sdk/auth.py

@@ -59,7 +59,7 @@ class GrantMethodBase:
         """Base payload."""
         return {
             "client_id": self.client_id,
-            "scope": "offline_access"
+            "scope": "openid"
         }
 
     def refresh_token(self, old_jwt: JWT) -> JWT:

dinamis_sdk/s3.py

@@ -26,8 +26,16 @@ from pystac_client import ItemSearch
 import packaging.version
 import pydantic
 
-from .utils import log, SIGNED_URL_TTL_MARGIN, CREDENTIALS, MAX_URLS, \
-    S3_SIGNING_ENDPOINT, S3_STORAGE_DOMAIN, SIGNED_URL_DURATION_SECONDS
+from .utils import (
+    log,
+    SIGNED_URL_TTL_MARGIN, 
+    CREDENTIALS, 
+    MAX_URLS,
+    S3_SIGNING_ENDPOINT, 
+    S3_STORAGE_DOMAIN, 
+    SIGNED_URL_DURATION_SECONDS,
+    BYPASS_API
+)
 
 _PYDANTIC_2_0 = packaging.version.parse(
     pydantic.__version__
@@ -35,6 +43,7 @@ _PYDANTIC_2_0 = packaging.version.parse(
 
 AssetLike = TypeVar("AssetLike", Asset, Dict[str, Any])
 
+# todo: fix the expression
 asset_xpr = re.compile(
     r"https://(?P<account>[A-z0-9]+?)"
     r"\.minio-dinamis\.apps\.okd\.crocc\.meso\.umontpellier\.fr/"
@@ -443,6 +452,8 @@ def get_signed_urls(
             "dinamis-secret-key": CREDENTIALS.secret_key
         })
         log.debug("Using credentials (access/secret keys)")
+    elif BYPASS_API:
+        log.debug("Using bypass API %s", BYPASS_API)
     else:
         from .auth import get_access_token
         access_token = get_access_token()

dinamis_sdk/utils.py

@@ -6,11 +6,16 @@ import appdirs
 from pydantic import BaseModel  # pylint: disable = no-name-in-module
 import requests
 
+# Env vars
+ENV_TTL_MARGIN = "DINAMIS_SDK_TTL_MARGIN"
+ENV_DURATION_SECS = "DINAMIS_SDK_DURATION_SECONDS"
+ENV_BYPASS_API = "DINAMIS_SDK_BYPASS_API"
+
 logging.basicConfig(level=os.environ.get("LOGLEVEL") or "INFO")
 log = logging.getLogger("dinamis_sdk")
 
 
-def _get_seconds(env_var_name: str, default: int) -> int:
+def _get_seconds(env_var_name: str, default: int = None) -> int:
     val = os.environ.get(env_var_name)
     if val:
         if val.isdigit():
@@ -24,11 +29,8 @@ def _get_seconds(env_var_name: str, default: int) -> int:
 
 
 # Signed TTL margin default to 1800 seconds (30 minutes), or env. var.
-SIGNED_URL_TTL_MARGIN = _get_seconds("DINAMIS_SDK_TTL_MARGIN", 1800)
-SIGNED_URL_DURATION_SECONDS = _get_seconds(
-    "DINAMIS_SDK_DURATION_SECONDS", 
-    None
-)
+SIGNED_URL_TTL_MARGIN = _get_seconds(ENV_TTL_MARGIN, 1800)
+SIGNED_URL_DURATION_SECONDS = _get_seconds(ENV_DURATION_SECS)
 
 MAX_URLS = 64
 S3_STORAGE_DOMAIN = "meso.umontpellier.fr"
@@ -83,12 +85,14 @@ def retrieve_token_endpoint(s3_signing_endpoint: str = S3_SIGNING_ENDPOINT):
     return oauth2_defs["flows"]["password"]["tokenUrl"]
 
 
+BYPASS_API = os.environ.get(ENV_BYPASS_API)
+
 # Token endpoint is typically something like: https://keycloak-dinamis.apps.okd
 # .crocc.meso.umontpellier.fr/auth/realms/dinamis/protocol/openid-connect/token
-TOKEN_ENDPOINT = retrieve_token_endpoint()
+TOKEN_ENDPOINT = None if BYPASS_API else retrieve_token_endpoint()
 # Auth base URL is typically something like: https://keycloak-dinamis.apps.okd.
 # crocc.meso.umontpellier.fr/auth/realms/dinamis/protocol/openid-connect
-AUTH_BASE_URL = TOKEN_ENDPOINT.rsplit('/', 1)[0]
+AUTH_BASE_URL = None if BYPASS_API else TOKEN_ENDPOINT.rsplit('/', 1)[0]
 
 # Token server (optional)
 TOKEN_SERVER = os.environ.get("DINAMIS_SDK_TOKEN_SERVER")

setup.py

@@ -12,7 +12,7 @@ install_requires = [
 
 setup(
     name="dinamis-sdk",
-    version="0.1.9",
+    version="0.1.10",
     description="DINAMIS SDK",
     python_requires=">=3.8",
     author="Remi Cresson",