add client credentials auth flow

README.md

@@ -37,6 +37,23 @@ export TRUSTIFY_URL="https://atlas.release.stage.devshift.net"
 export AUTH_ENDPOINT="https://auth.stage.redhat.com/auth/realms/EmployeeIDP/protocol/openid-connect"
 ```
 
+TPA / Amazon Cognito (e.g. Red Hat Trusted Profile Analyzer):
+
+Use your Cognito domain, CLI client ID, and client secret. The **client credentials** flow requires no browser or callback URL (unlike the authorization code flow):
+
+```bash
+export TRUSTIFY_URL="https://server-tpa221on420.apps.ps420.tpa.rhocf-dev.net"
+export AUTH_ENDPOINT="https://tpa221on420.auth.eu-west-1.amazoncognito.com/oauth2"
+export OIDC_CLIENT_ID="1g3hp92dqfrjph3om08g2is0bo"
+export OIDC_CLIENT_SECRET="your-client-secret"
+```
+
+The Cognito app client must have the **client credentials** grant enabled (not authorization code). Run `trust-purl` or `trust-products`; authentication happens automatically via `client_id` and `client_secret`.
+
+**Security:** Don't commit or share `OIDC_CLIENT_SECRET`. Use environment variables or a secrets manager.
+
+If the app client uses the authorization code flow instead (e.g. browser login), add `export OIDC_AUTH_PATH="authorize"` (and have the admin add `http://localhost:8650/index.html` to Cognito's allowed callbacks).
+
 For local or public Trustify instances without authentication, simply omit `AUTH_ENDPOINT`:
 
 ```bash
@@ -55,6 +72,11 @@ Optional Configuration:
 ```bash
 # Set custom configuration directory (defaults to ~/.config/trustshell/)
 export TRUSTSHELL_SCRATCH="/path/to/custom/config/dir"
+
+# OIDC overrides for non-Keycloak providers (e.g. Amazon Cognito):
+# OIDC_CLIENT_ID - app client ID
+# OIDC_CLIENT_SECRET - enables client_credentials flow (no browser/callback) when set
+# OIDC_AUTH_PATH - auth path for PKCE (Keycloak: "auth", Cognito: "authorize")
 ```
 
 ### Running in a container

src/trustshell/__init__.py

@@ -24,6 +24,7 @@
     build_url,
     code_to_token,
     gen_things,
+    get_client_credentials_token,
 )
 
 CONFIG_DIR = os.path.expanduser(
@@ -270,6 +271,9 @@ def log_message(self, format: str, *args: Any) -> None:
 
 
 def get_access_token() -> str:
+    # Client credentials flow: no browser or callback URL required
+    if cc_token := get_client_credentials_token():
+        return cc_token
     if HEADLESS or LOCAL_AUTH_SERVER_PORT:
         logger.debug(
             f"Running in HEADLESS mode, trying OIDC PKCE flow with {REDIRECT_URI}"

src/trustshell/oidc/__init__.py

@@ -8,6 +8,7 @@
     build_url,
     code_to_token,
     get_fresh_token,
+    get_client_credentials_token,
     LOCAL_SERVER_PORT,
     REDIRECT_URI,
     AUTH_ENDPOINT,
@@ -18,6 +19,7 @@
     "build_url",
     "code_to_token",
     "get_fresh_token",
+    "get_client_credentials_token",
     "LOCAL_SERVER_PORT",
     "REDIRECT_URI",
     "AUTH_ENDPOINT",

src/trustshell/oidc/oidc_pkce_authcode.py

@@ -12,7 +12,12 @@
 import httpx
 
 logger = logging.getLogger("trustshell")
-CLIENT_ID = "atlas-frontend"
+# Client ID - Keycloak uses "atlas-frontend"; Cognito TPA uses a dedicated CLI client
+CLIENT_ID = os.getenv("OIDC_CLIENT_ID", "atlas-frontend")
+# Client secret - required for Cognito confidential clients during token exchange
+CLIENT_SECRET = os.getenv("OIDC_CLIENT_SECRET", "")
+# Auth path: Keycloak uses "auth", Amazon Cognito uses "authorize"
+OIDC_AUTH_PATH = os.getenv("OIDC_AUTH_PATH", "auth")
 # this script will spawn an HTTP server to capture the code your browser gets from the SSO server
 LOCAL_SERVER_PORT = 8650
 REDIRECT_URI = "http://localhost:" + str(LOCAL_SERVER_PORT) + "/index.html"
@@ -28,6 +33,33 @@
 token_endpoint = f"{AUTH_ENDPOINT}/token"
 
 
+def get_client_credentials_token() -> str | None:
+    """
+    Obtain an access token via OAuth2 client_credentials flow.
+    No browser or callback URL required - uses only client_id and client_secret.
+    Returns None if client credentials are not configured or the grant is unsupported.
+    """
+    if not CLIENT_SECRET or not AUTH_ENDPOINT:
+        return None
+    logger.debug("Fetching token via client_credentials flow")
+    data = {
+        "grant_type": "client_credentials",
+        "client_id": CLIENT_ID,
+        "client_secret": CLIENT_SECRET,
+    }
+    resp = httpx.post(url=token_endpoint, data=data)
+    if resp.is_error:
+        logger.debug(
+            "Client credentials flow failed: [%s] %s", resp.status_code, resp.text
+        )
+        return None
+    token_data = resp.json()
+    if "access_token" not in token_data:
+        logger.debug("No access_token in response: %s", token_data)
+        return None
+    return str(token_data["access_token"])
+
+
 def gen_things() -> tuple[str, str, str]:
     logging.debug("Generating verifier, challenge, state")
     code_verifier, code_challenge = pkce.generate_pkce_pair()
@@ -49,9 +81,8 @@ def build_url(code_challenge: str, state: str, auth_server: str = "") -> str:
         "state": state,
     }
     encoded_params = urllib.parse.urlencode(params)
-    authz_endpoint = f"{AUTH_ENDPOINT}/auth"
-    if auth_server:
-        authz_endpoint = f"{auth_server}/auth"
+    base = auth_server if auth_server else AUTH_ENDPOINT
+    authz_endpoint = f"{base}/{OIDC_AUTH_PATH}"
     url = f"{authz_endpoint}?{encoded_params}"
     return url
 
@@ -60,13 +91,15 @@ def code_to_token(code: str, code_verifier: str) -> tuple[str, str, str]:
     logger.debug(
         "Exchanging the code for a token via http calls inside of this script."
     )
-    data = {
+    data: dict[str, str] = {
         "grant_type": GRANT_TYPE,
         "client_id": CLIENT_ID,
         "code_verifier": code_verifier,
         "code": code,
         "redirect_uri": REDIRECT_URI,
     }
+    if CLIENT_SECRET:
+        data["client_secret"] = CLIENT_SECRET
     c2t = httpx.post(url=token_endpoint, data=data)
     c2t_json = json.loads(c2t.text)
     access_token = c2t_json["access_token"]
@@ -107,11 +140,13 @@ def get_fresh_token(refresh_token: str) -> tuple[str, str]:
     logger.debug(
         "Exchange the refresh token for a new access token via http calls inside of this script."
     )
-    data = {
+    data: dict[str, str] = {
         "grant_type": "refresh_token",
         "client_id": CLIENT_ID,
         "refresh_token": refresh_token,
     }
+    if CLIENT_SECRET:
+        data["client_secret"] = CLIENT_SECRET
     r2a = httpx.post(url=token_endpoint, data=data)
     r2a_json = json.loads(r2a.text)
     access_token = r2a_json["access_token"]