Change auth from api-key to client credentials

canvas_cli/apps/auth/auth.py

@@ -1,10 +1,17 @@
-from urllib.parse import urlparse
+from typing import Optional
+from urllib.parse import parse_qs, urlparse
 
 import typer
 
-from canvas_cli.apps.auth.utils import delete_password, get_password, set_password
+from canvas_cli.apps.auth.utils import (
+    delete_password,
+    get_or_request_api_token,
+    get_password,
+    set_password,
+)
 from canvas_cli.utils.context.context import context
 from canvas_cli.utils.print import print
+from canvas_cli.utils.validators import get_default_host
 
 app = typer.Typer()
 
@@ -26,23 +33,26 @@ def validate_host(host: str) -> str:
 
 
 @app.command(
-    short_help="Add a host=api-key pair to the keychain, so it can be used in other requests"
+    short_help="Adds host=client_id&client_secret pair to the keychain, so it can be used to request future tokens"
 )
-def add_api_key(
+def add_api_client_credentials(
     host: str = typer.Option(..., prompt=True, callback=validate_host),
-    api_key: str = typer.Option(..., prompt=True),
+    client_id: str = typer.Option(..., prompt=True),
+    client_secret: str = typer.Option(..., prompt=True),
     is_default: bool = typer.Option(..., prompt=True),
 ) -> None:
-    """Add a host=api-key pair to the keychain, so it can be used in other requests.
+    """Adds host=client_id and host=client_secret pair to the keychain, so it can be used to request future tokens.
     Optionally set a default so `--host` isn't required everywhere
     """
-    print.verbose(f"Saving an api key for {host}...")
+    print.verbose(f"Saving a client_id and client_secret for {host}...")
 
-    current_credential = get_password(host)
-    if current_credential:
-        typer.confirm(f"An api-key already exists for {host}, override?", abort=True)
+    current_creds = get_password(host)
+    if current_creds:
+        typer.confirm(
+            f"An client_id and client_secret already exist for {host}, override?", abort=True
+        )
 
-    set_password(username=host, password=api_key)
+    set_password(username=host, password=f"client_id={client_id}&client_secret={client_secret}")
 
     if is_default:
         context.default_host = host
@@ -52,7 +62,7 @@ def add_api_key(
 
 
 @app.command(short_help="Removes a host from the keychain, and as the default if it's the one")
-def remove_api_key(host: str = typer.Argument(..., callback=validate_host)) -> None:
+def remove_api_client_credentials(host: str = typer.Argument(..., callback=validate_host)) -> None:
     """Removes a host from the keychain, and as the default if it's the one.
     This method always succeeds, regardless of username existence.
     """
@@ -67,13 +77,14 @@ def remove_api_key(host: str = typer.Argument(..., callback=validate_host)) -> N
     print(f"{host} removed")
 
 
-@app.command(short_help="Print the api_key for the given host")
-def get_api_key(host: str = typer.Argument(..., callback=validate_host)) -> None:
-    """Print the api_key for the given host."""
-    api_key = get_password(host)
+@app.command(short_help="Print the api_client_credentials for the given host")
+def get_api_client_credentials(host: str = typer.Argument(..., callback=validate_host)) -> None:
+    """Print the api_client_credentials for the given host."""
+    api_client_credentials = get_password(host)
 
-    if api_key:
-        print.json(message=None, host=host, api_key=api_key)
+    if api_client_credentials:
+        creds = parse_qs(api_client_credentials)
+        print.json(message=None, host=host, **creds)
     else:
         print.json(f"{host} not found.", success=False)
 
@@ -83,10 +94,22 @@ def set_default_host(host: str = typer.Argument(..., callback=validate_host)) ->
     """Set the host as the default host in the config file. Validates it exists in the keychain."""
     print.verbose(f"Setting {host} as default")
 
-    api_key = get_password(host)
-
-    if not api_key:
+    if not get_password(host):
         print.json(f"{host} doesn't exist in the keychain. Please add it first.", success=False)
         raise typer.Abort()
 
     context.default_host = host
+
+
+@app.command(short_help="Print the current api_token for the given host.")
+def get_api_token(
+    host: Optional[str] = typer.Option(callback=get_default_host, default=None),
+    client_id: Optional[str] = typer.Option(default=None),
+    client_secret: Optional[str] = typer.Option(default=None),
+) -> None:
+    """Print the current api_token for the given host."""
+    if not host:
+        raise typer.BadParameter("Please specify a host or set a default via the `auth` command")
+
+    token = get_or_request_api_token(host, client_id, client_secret)
+    print.json(message=None, token=token)

canvas_cli/apps/auth/tests.py

@@ -51,39 +51,44 @@ def test_validate_host_valid(scheme: str) -> None:
 
 @patch("keyring.get_password")
 @patch("keyring.set_password")
-def test_add_api_key_new_host(mock_set_password: MagicMock, mock_get_password: MagicMock) -> None:
+def test_add_api_client_credentials_new_host(
+    mock_set_password: MagicMock, mock_get_password: MagicMock
+) -> None:
     """Test a new password is set if the host doesn't already exist and ensure it isn't written to the config."""
     mock_get_password.return_value = None
     runner.invoke(
         app,
-        "auth add-api-key --host localhost --api-key mock-api-key --no-is-default",
+        "auth add-api-client-credentials --host localhost --client-id mock-client-id --client-secret mock-client-secret --no-is-default",
     )
     mock_set_password.assert_called()
     assert context.default_host is None
 
 
 @patch("keyring.get_password")
 @patch("keyring.set_password")
-def test_add_api_key_new_host_set_default(
+def test_add_api_client_credentials_new_host_set_default(
     mock_set_password: MagicMock, mock_get_password: MagicMock
 ) -> None:
     """Test a new password is set if the host doesn't already exist and ensure it is written to the config."""
     mock_get_password.return_value = None
-    runner.invoke(app, "auth add-api-key --host localhost --api-key mock-api-key --is-default")
+    runner.invoke(
+        app,
+        "auth add-api-client-credentials --host localhost --client-id mock-client-id --client-secret mock-client-secret --is-default",
+    )
     mock_set_password.assert_called()
     assert context.default_host == "http://localhost"
 
 
 @patch("keyring.get_password")
 @patch("keyring.set_password")
-def test_add_api_key_existing_host_overwrite(
+def test_add_api_client_credentials_existing_host_overwrite(
     mock_set_password: MagicMock, mock_get_password: MagicMock
 ) -> None:
     """Test a new password is set if the host exists, but we want to override."""
-    mock_get_password.return_value = "a-domain-name.com"
+    mock_get_password.return_value = "http://a-domain-name.com"
     result = runner.invoke(
         app,
-        "auth add-api-key --host http://a-domain-name.com --api-key mock-api-key --no-is-default",
+        "auth add-api-client-credentials --host http://a-domain-name.com --client-id mock-client-id --client-secret mock-client-secret --no-is-default",
         input="y\n",
     )
     mock_set_password.assert_called()
@@ -96,11 +101,11 @@ def test_add_api_key_existing_host_abort(
     mock_set_password: MagicMock, mock_get_password: MagicMock
 ) -> None:
     """Test a new password is NOT set if the host already exists and we DO NOT override."""
-    mock_get_password.return_value = "a-domain-name.com"
+    mock_get_password.return_value = "http://a-domain-name.com"
 
     result = runner.invoke(
         app,
-        "auth add-api-key --host http://a-domain-name.com --api-key mock-api-key --no-is-default",
+        "auth add-api-client-credentials --host http://a-domain-name.com --client-id mock-client-id --client-secret mock-client-secret --no-is-default",
         input="n\n",
     )
 
@@ -119,10 +124,13 @@ def test_remove_api_key(
     """Test removing an api-key deletes it from the config file."""
     # First we need to add an api-key to the config file
     # The following command is tested, so we're not asserting the result
-    runner.invoke(app, "auth add-api-key --host localhost --api-key mock-api-key --is-default")
+    runner.invoke(
+        app,
+        "auth add-api-client-credentials --host localhost --client-id mock-client-id --client-secret mock-client-secret --is-default",
+    )
 
     # Then we remove the same host, which should delete the entry from the config file
-    result = runner.invoke(app, "auth remove-api-key localhost")
+    result = runner.invoke(app, "auth remove-api-client-credentials localhost")
 
     assert context.default_host is None
     mock_delete_password.assert_called()

canvas_cli/apps/auth/utils.py

@@ -1,4 +1,9 @@
+from datetime import datetime, timedelta
+
 import keyring
+import requests
+
+from canvas_cli.utils.context.context import context
 
 # Keyring namespace we'll use
 KEYRING_SERVICE = __name__
@@ -17,3 +22,59 @@ def set_password(username: str, password: str) -> None:
 def delete_password(username: str) -> None:
     """Delete the password for the given username."""
     keyring.delete_password(KEYRING_SERVICE, username=username)
+
+
+def get_api_client_credentials(
+    host: str, client_id: str | None, client_secret: str | None
+) -> str | None:
+    """Either return the given api_key, or fetch it from the keyring."""
+    if client_id and client_secret:
+        return f"client_id={client_id}&client_secret={client_secret}"
+
+    return get_password(host)
+
+
+def request_api_token(host: str, api_client_credentials: str) -> str | None:
+    """Request an api token using the provided client_id and client_secret."""
+    token_response = requests.post(
+        f"{host}/auth/token/",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        data=f"grant_type=client_credentials&{api_client_credentials}",
+    )
+    if token_response.status_code != requests.codes.ok:
+        raise Exception(
+            "Unable to get a valid access token. Please check your host, client_id, and client_secret"
+        )
+
+    token_response_json = token_response.json()
+
+    token_expiration_date = datetime.now() + timedelta(seconds=token_response_json["expires_in"])
+    context.token_expiration_date = token_expiration_date.isoformat()
+    return token_response_json["access_token"]
+
+
+def is_token_valid() -> bool:
+    """True if the token has not expired yet."""
+    expiration_date = context.token_expiration_date
+    return expiration_date is not None and datetime.fromisoformat(expiration_date) > datetime.now()
+
+
+def get_or_request_api_token(host: str, client_id: str | None, client_secret: str | None) -> str:
+    """Returns an existing stored token if it has not expired, or requests a new one."""
+    host_token_key = f"{host}|token"
+    token = get_password(host_token_key)
+
+    if token and is_token_valid():
+        return token
+
+    if not (api_client_credentials := get_api_client_credentials(host, client_id, client_secret)):
+        raise Exception(
+            "Please specify a client_id and client_secret or add them via the `auth` command"
+        )
+
+    if not (new_token := request_api_token(host, api_client_credentials)):
+        raise Exception(
+            "A token could not be acquired from the given host, client_id, and client_secret"
+        )
+    set_password(host_token_key, new_token)
+    return new_token

canvas_cli/apps/logs/logs.py

@@ -4,8 +4,9 @@
 import typer
 import websocket
 
+from canvas_cli.apps.auth.utils import get_or_request_api_token
 from canvas_cli.utils.print import print
-from canvas_cli.utils.validators import get_api_key, get_default_host
+from canvas_cli.utils.validators import get_default_host
 
 
 def _on_message(ws: websocket.WebSocketApp, message: str) -> None:
@@ -28,16 +29,18 @@ def logs(
     host: Optional[str] = typer.Option(
         callback=get_default_host, help="Canvas instance to connect to", default=None
     ),
-    api_key: Optional[str] = typer.Option(
-        help="Canvas api-key for the provided host", default=None
+    client_id: Optional[str] = typer.Option(
+        help="Canvas client_id for the provided host", default=None
+    ),
+    client_secret: Optional[str] = typer.Option(
+        help="Canvas client_secret for the provided host", default=None
     ),
 ) -> None:
     """Listens and prints log streams from the instance."""
     if not host:
         raise typer.BadParameter("Please specify a host or set a default via the `auth` command")
 
-    if not (final_api_key := get_api_key(host, api_key)):
-        raise typer.BadParameter("Please specify an api-key or add one via the `auth` command")
+    token = get_or_request_api_token(host, client_id, client_secret)
 
     # Resolve the instance name from the Canvas host URL (e.g., extract
     # 'example' from 'https://example.canvasmedical.com/')
@@ -47,7 +50,7 @@ def logs(
     print.json(
         f"Connecting to the log stream. Please be patient as there may be a delay before log messages appear."
     )
-    websocket_uri = f"wss://logs.console.canvasmedical.com/{instance}?token={final_api_key}"
+    websocket_uri = f"wss://logs.console.canvasmedical.com/{instance}?token={token}"
 
     try:
         ws = websocket.WebSocketApp(