Add OAuth2 scopes support for token requests

README.md

@@ -17,6 +17,26 @@ The credentials will contain the Client Id and Client Secret.
 4. Set to credentials tuple `auth=(client_id, client_secret)`
 5. Store credentials in the system keyring (see [Storing credentials in the keyring](#storing-credentials-in-the-keyring))
 
+OAuth2 scopes can be configured to restrict the permissions of tokens. Scopes are resolved
+in this order (first match wins):
+1. Explicit `scopes` parameter passed to client constructors or `--scopes` on the CLI
+2. `scopes` field in the environment configuration (`environments.json`)
+3. `scopes` field in the credentials file
+
+The credentials file supports an optional `scopes` field — an array of OAuth2 scopes
+to request when fetching tokens:
+
+```json
+{
+  "clientId": "...",
+  "clientSecret": "...",
+  "email": "...",
+  "userId": 123,
+  "issuer": "...",
+  "scopes": ["api:read", "api:write"]
+}
+```
+
 API clients such as the `InputApiClient` accept this `auth` parameter.
 
 Under the hood, they commonly use the AuthSession class which is implements a `requests` session with automatic token
@@ -55,7 +75,8 @@ The CLI can be configured with a JSON file at `~/.config/kognic/environments.jso
     "example": {
       "host": "example.kognic.com",
       "auth_server": "https://auth.example.kognic.com",
-      "credentials": "~/.config/kognic/credentials-example.json"
+      "credentials": "~/.config/kognic/credentials-example.json",
+      "scopes": ["api:read", "api:write"]
     }
   }
 }
@@ -64,6 +85,7 @@ The CLI can be configured with a JSON file at `~/.config/kognic/environments.jso
 Each environment has the following fields:
 - `host` - The API hostname, used by `kog` to automatically match an environment based on the request URL.
 - `auth_server` - The OAuth server URL used to fetch tokens.
+- `scopes` *(optional)* - OAuth2 scopes to request when fetching tokens. Overrides scopes from the credentials file.
 - `credentials` *(optional)* - Where to load credentials from. Three formats are supported:
   - A file path: `"~/.config/kognic/credentials-prod.json"` (tilde `~` is expanded)
   - A keyring reference: `"keyring://production"` (loads from the system keyring under the named profile)
@@ -76,14 +98,15 @@ Each environment has the following fields:
 Generate an access token for Kognic API authentication.
 
 ```bash
-kognic-auth get-access-token [--server SERVER] [--credentials FILE] [--env NAME] [--env-config-file-path FILE]
+kognic-auth get-access-token [--server SERVER] [--credentials FILE] [--env NAME] [--env-config-file-path FILE] [--scopes SCOPE ...]
 ```
 
 **Options:**
 - `--server` - Authentication server URL (default: `https://auth.app.kognic.com`)
 - `--credentials` - Path to JSON credentials file. If not provided, credentials are read from environment variables.
 - `--env` - Use a named environment from the config file.
 - `--env-config-file-path` - Environment config file path (default: `~/.config/kognic/environments.json`)
+- `--scopes` - OAuth2 scopes to request (e.g. `--scopes api:read api:write`). Overrides scopes from environment config and credentials file.
 
 When `--env` is provided, the auth server and credentials are resolved from the config file. Explicit `--server` or `--credentials` flags override the environment values.
 
@@ -100,13 +123,16 @@ kognic-auth get-access-token --env example
 
 # Using an environment but overriding the server
 kognic-auth get-access-token --env example --server https://custom.server
+
+# Request specific scopes
+kognic-auth get-access-token --scopes api:read api:write
 ```
 
 ### kognic-auth credentials
 
-Manage credentials stored in the system keyring (macOS Keychain, GNOME Keyring, Windows Credential Manager, etc.). 
+Manage credentials stored in the system keyring (macOS Keychain, GNOME Keyring, Windows Credential Manager, etc.).
 This is the recommended way to store credentials on a developer machine — more secure than a credentials file and no environment variables in shell profiles.
-Credentials files downloaded from the Kognic Platform UI can be put into the keyring. 
+Credentials files downloaded from the Kognic Platform UI can be put into the keyring.
 
 ```bash
 kognic-auth credentials put FILE [--env ENV]
@@ -271,6 +297,9 @@ client = MyApiClient(auth=("my-client-id", "my-client-secret"))
 
 # Or with credentials file
 client = MyApiClient(auth="~/.config/kognic/credentials.json")
+
+# With explicit scopes
+client = MyApiClient(auth=("my-client-id", "my-client-secret"), scopes=["api:read", "api:write"])
 ```
 
 ### Async Client (httpx)
@@ -287,6 +316,10 @@ class MyAsyncApiClient(BaseAsyncApiClient):
 # Usage as async context manager
 async with MyAsyncApiClient() as client:
     resource = await client.get_resource("123")
+
+# With explicit scopes
+async with MyAsyncApiClient(scopes=["api:read"]) as client:
+    resource = await client.get_resource("123")
 ```
 
 ## Serialization & Deserialization

src/kognic/auth/cli/get_access_token.py

@@ -43,6 +43,12 @@ def register_parser(subparsers: argparse._SubParsersAction) -> argparse.Argument
         help="Token cache backend: auto (default), keyring, file, or none. "
         "Auto will use keyring if available, otherwise file-based caching.",
     )
+    token_parser.add_argument(
+        "--scopes",
+        nargs="+",
+        default=None,
+        help="OAuth2 scopes to request, e.g. --scopes api:read api:write",
+    )
 
     return token_parser
 
@@ -51,6 +57,7 @@ def run(parsed: argparse.Namespace) -> int:
     try:
         host = parsed.server
         credentials = parsed.credentials
+        env_scopes = []
 
         if parsed.env_name:
             config = load_kognic_env_config(parsed.env_config_file_path)
@@ -63,13 +70,19 @@ def run(parsed: argparse.Namespace) -> int:
                 host = ctx.auth_server
             if credentials is None:
                 credentials = ctx.credentials
+            env_scopes = ctx.scopes
 
         auth_host = host or DEFAULT_HOST
 
+        scopes = parsed.scopes
+        if scopes is None and env_scopes:
+            scopes = env_scopes
+
         provider = make_token_provider(
             auth=credentials,
             auth_host=auth_host,
             token_cache=make_cache(parsed.token_cache),
+            scopes=scopes,
         )
         print(provider.ensure_token()["access_token"])
         return 0

src/kognic/auth/credentials.py

@@ -1,6 +1,6 @@
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import datetime
-from typing import Optional
+from typing import List, Optional
 
 
 @dataclass
@@ -13,3 +13,4 @@ class ApiCredentials:
     name: str = "API Credentials"
     created: Optional[datetime] = None
     expires: Optional[datetime] = None
+    scopes: List[str] = field(default_factory=list)

src/kognic/auth/credentials_parser.py

@@ -73,6 +73,7 @@ def parse_credentials(path: Union[str, os.PathLike, dict]) -> ApiCredentials:
         name=credentials.get("name", "API Credentials"),
         created=_parse_optional_datetime(credentials.get("created")),
         expires=_parse_optional_datetime(credentials.get("expires")),
+        scopes=credentials.get("scopes", []),
     )
 
 

src/kognic/auth/env_config.py

@@ -2,7 +2,7 @@
 import os
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Optional, Union
+from typing import List, Optional, Union
 from urllib.parse import urlparse
 
 from kognic.auth import DEFAULT_ENV_CONFIG_FILE_PATH, DEFAULT_HOST, DEFAULT_KOGNIC_PLATFORM
@@ -14,6 +14,7 @@ class Environment:
     host: str
     auth_server: str
     credentials: Optional[str] = None
+    scopes: List[str] = field(default_factory=list)
 
 
 @dataclass
@@ -40,6 +41,7 @@ def load_kognic_env_config(path: Union[str, os.PathLike] = DEFAULT_ENV_CONFIG_FI
             host=env_data["host"],
             auth_server=env_data["auth_server"],
             credentials=credentials,
+            scopes=env_data.get("scopes", []),
         )
 
     return KognicEnvConfig(

src/kognic/auth/httpx/async_client.py

@@ -1,5 +1,6 @@
 import logging
 from asyncio import Lock
+from typing import List, Optional
 
 import httpx
 from authlib.integrations.httpx_client import AsyncOAuth2Client
@@ -30,6 +31,7 @@ def __init__(
         auth=None,
         host: str = DEFAULT_HOST,
         token_endpoint: str = DEFAULT_TOKEN_ENDPOINT_RELPATH,
+        scopes: Optional[List[str]] = None,
         **kwargs,
     ):
         """Initialize the async auth client.
@@ -38,6 +40,7 @@ def __init__(
             auth: Authentication credentials - path to credentials file or (client_id, client_secret) tuple
             host: Base url for authentication server
             token_endpoint: Relative path to the token endpoint
+            scopes: OAuth2 scopes to request, e.g. ["api:read", "api:write"].
             **kwargs: Additional params to pass into Httpx Client Constructor
         """
         self.host = host
@@ -50,12 +53,16 @@ def __init__(
         client_id = creds.client_id if creds else None
         client_secret = creds.client_secret if creds else None
 
+        if scopes is None and creds and creds.scopes:
+            scopes = creds.scopes
+
         self._oauth_client = _AsyncFixedClient(
             client_id=client_id,
             client_secret=client_secret,
             update_token=self._update_token,
             token_endpoint=self.token_url,
             grant_type="client_credentials",
+            scope=" ".join(scopes) if scopes else None,
             **kwargs,
         )
         self._oauth_client.register_compliance_hook("access_token_response", AuthClient.check_rate_limit)

src/kognic/auth/httpx/base_client.py

@@ -5,7 +5,7 @@
 import asyncio
 import logging
 import os
-from typing import TYPE_CHECKING, Any, Callable, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, List, Optional, Union
 
 from kognic.auth.credentials_parser import ANY_AUTH_TYPE
 
@@ -71,6 +71,7 @@ def __init__(
         client_name: Optional[str] = "auto",
         json_serializer: Callable[[Any], Any] = serialize_body,
         sunset_handler: Optional[SunsetHandler] = _DEFAULT_SUNSET_HANDLER,
+        scopes: Optional[List[str]] = None,
         **kwargs,
     ):
         """Initialize the async API client.
@@ -83,6 +84,7 @@ def __init__(
             json_serializer: Callable to serialize request bodies. Defaults to serialize_body.
             sunset_handler: Callable invoked with ``(sunset_date, method, url)`` when a sunset header
                 is detected. Defaults to logging a warning or error. Pass ``None`` to disable.
+            scopes: OAuth2 scopes to request, e.g. ["api:read", "api:write"].
             **kwargs: Additional arguments passed to the underlying httpx client (e.g. timeout, verify).
         """
         if client_name == "auto":
@@ -98,6 +100,7 @@ def __init__(
             auth=auth,
             host=auth_host,
             token_endpoint=auth_token_endpoint,
+            scopes=scopes,
             headers=headers,
             **kwargs,
         )
@@ -166,4 +169,6 @@ def from_env(
         resolved = cfg.environments[env]
         kwargs.setdefault("auth", resolved.credentials)
         kwargs["auth_host"] = resolved.auth_server
+        if resolved.scopes and "scopes" not in kwargs:
+            kwargs["scopes"] = resolved.scopes
         return cls(**kwargs)

src/kognic/auth/internal/token_cache/_base.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import time
 from abc import ABC, abstractmethod
 from typing import Optional
@@ -11,8 +12,12 @@
 EXPIRY_MARGIN_SECONDS = 30
 
 
-def make_key(auth_server: str, client_id: str) -> str:
-    return f"{auth_server}:{client_id}"
+def make_key(auth_server: str, client_id: str, scopes: Optional[str] = None) -> str:
+    key = f"{auth_server}:{client_id}"
+    if scopes:
+        scope_hash = hashlib.sha256(scopes.encode()).hexdigest()[:12]
+        key = f"{key}:s-{scope_hash}"
+    return key
 
 
 def is_valid(token: dict) -> bool:
@@ -26,13 +31,13 @@ class TokenCache(ABC):
     """Abstract base class for token caches."""
 
     @abstractmethod
-    def load(self, auth_server: str, client_id: str) -> Optional[dict]:
+    def load(self, auth_server: str, client_id: str, scopes: Optional[str] = None) -> Optional[dict]:
         """Return a non-expired token dict, or None."""
 
     @abstractmethod
-    def save(self, auth_server: str, client_id: str, token: dict) -> None:
+    def save(self, auth_server: str, client_id: str, token: dict, scopes: Optional[str] = None) -> None:
         """Persist a token dict. Silently ignores errors."""
 
     @abstractmethod
-    def clear(self, auth_server: str, client_id: str) -> None:
+    def clear(self, auth_server: str, client_id: str, scopes: Optional[str] = None) -> None:
         """Remove a cached token. Silently ignores errors."""

src/kognic/auth/internal/token_cache/_file.py

@@ -30,9 +30,9 @@ def _save_all(self, data: dict) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)
         self.path.write_text(json.dumps(data, indent=2))
 
-    def load(self, auth_server: str, client_id: str) -> Optional[dict]:
+    def load(self, auth_server: str, client_id: str, scopes: Optional[str] = None) -> Optional[dict]:
         try:
-            key = make_key(auth_server, client_id)
+            key = make_key(auth_server, client_id, scopes)
             token = self._load_all().get(key)
             if token is None:
                 return None
@@ -45,19 +45,19 @@ def load(self, auth_server: str, client_id: str) -> Optional[dict]:
             log.debug("Failed to load token from file cache", exc_info=True)
             return None
 
-    def save(self, auth_server: str, client_id: str, token: dict) -> None:
+    def save(self, auth_server: str, client_id: str, token: dict, scopes: Optional[str] = None) -> None:
         try:
-            key = make_key(auth_server, client_id)
+            key = make_key(auth_server, client_id, scopes)
             data = self._load_all()
             data[key] = token
             self._save_all(data)
             log.debug("Saved token to file cache for key=%s", key)
         except Exception:
             log.debug("Failed to save token to file cache", exc_info=True)
 
-    def clear(self, auth_server: str, client_id: str) -> None:
+    def clear(self, auth_server: str, client_id: str, scopes: Optional[str] = None) -> None:
         try:
-            key = make_key(auth_server, client_id)
+            key = make_key(auth_server, client_id, scopes)
             data = self._load_all()
             if key in data:
                 del data[key]

src/kognic/auth/internal/token_cache/_keyring.py

@@ -35,12 +35,12 @@ def _keyring(self):
             return None
         return self._keyring_module
 
-    def load(self, auth_server: str, client_id: str) -> Optional[dict]:
+    def load(self, auth_server: str, client_id: str, scopes: Optional[str] = None) -> Optional[dict]:
         kr = self._keyring()
         if kr is None:
             return None
         try:
-            key = make_key(auth_server, client_id)
+            key = make_key(auth_server, client_id, scopes)
             stored = kr.get_password(SERVICE_NAME, key)
             if stored is None:
                 return None
@@ -54,23 +54,23 @@ def load(self, auth_server: str, client_id: str) -> Optional[dict]:
             log.debug("Failed to load token from keyring", exc_info=True)
             return None
 
-    def save(self, auth_server: str, client_id: str, token: dict) -> None:
+    def save(self, auth_server: str, client_id: str, token: dict, scopes: Optional[str] = None) -> None:
         kr = self._keyring()
         if kr is None:
             return
         try:
-            key = make_key(auth_server, client_id)
+            key = make_key(auth_server, client_id, scopes)
             kr.set_password(SERVICE_NAME, key, json.dumps(token))
             log.debug("Saved token to keyring for key=%s", key)
         except Exception:
             log.debug("Failed to save token to keyring", exc_info=True)
 
-    def clear(self, auth_server: str, client_id: str) -> None:
+    def clear(self, auth_server: str, client_id: str, scopes: Optional[str] = None) -> None:
         kr = self._keyring()
         if kr is None:
             return
         try:
-            key = make_key(auth_server, client_id)
+            key = make_key(auth_server, client_id, scopes)
             kr.delete_password(SERVICE_NAME, key)
             log.debug("Cleared cached token from keyring for key=%s", key)
         except Exception: