globus api --scope-string (for clients)

changelog.d/20231206_130249_derek_api_scopes.md

@@ -0,0 +1,12 @@
+
+### Enhancements
+
+* The ``globus api <service>`` command now supports a ``--scope-string`` parameter.
+
+  * If supplied, the CLI will enforce that any specified scope strings are included
+    in consent requirements *in addition to* standard service scope requirements.
+
+  * This parameter may be supplied multiple times to specify multiple scope strings.
+
+  * This parameter is only supported in the context of Client Credentials-based authentication.
+    ([Client Credentials with GLOBUS_CLI_CLIENT_ID](https://docs.globus.org/cli/environment_variables/#client_credentials_with_globus_cli_client_id))

src/globus_cli/commands/api.py

@@ -9,7 +9,7 @@
 import globus_sdk
 
 from globus_cli import termio, version
-from globus_cli.login_manager import LoginManager
+from globus_cli.login_manager import LoginManager, is_client_login
 from globus_cli.login_manager.scopes import CLI_SCOPE_REQUIREMENTS
 from globus_cli.parsing import command, group, mutex_option_group
 from globus_cli.termio import display
@@ -124,6 +124,20 @@ def print_error_or_response(
         display(data, simple_text=data.text)
 
 
+def _get_resource_server(service_name: str) -> str:
+    _resource_server = {
+        "auth": globus_sdk.AuthClient.resource_server,
+        "flows": globus_sdk.FlowsClient.resource_server,
+        "groups": globus_sdk.GroupsClient.resource_server,
+        "search": globus_sdk.SearchClient.resource_server,
+        "transfer": globus_sdk.TransferClient.resource_server,
+        "timer": globus_sdk.TimerClient.resource_server,
+    }.get(service_name)
+    if _resource_server is None:
+        raise NotImplementedError(f"unrecognized service: {service_name}")
+    return _resource_server
+
+
 def _get_client(
     login_manager: LoginManager, service_name: str
 ) -> globus_sdk.BaseClient:
@@ -240,6 +254,16 @@ def build_command(service_name: ServiceNameLiteral) -> click.Command:
         ),
     )
     @click.option("--no-retry", is_flag=True, help="Disable built-in request retries")
+    @click.option(
+        "--scope-string",
+        type=str,
+        multiple=True,
+        help=(
+            "A scope string to consent to for this command (only supported for "
+            "confidential client usage). "
+            "Pass this option multiple times to specify multiple scopes."
+        ),
+    )
     @mutex_option_group("--body", "--body-file")
     def service_command(
         login_manager: LoginManager,
@@ -254,6 +278,7 @@ def service_command(
         allow_errors: bool,
         allow_redirects: bool,
         no_retry: bool,
+        scope_string: tuple[str, ...],
     ) -> None:
         # the overall flow of this command will be as follows:
         # - prepare a client
@@ -264,6 +289,15 @@ def service_command(
         #   - on success or error with --allow-errors, print
         #   - on error without --allow-errors, reraise
 
+        if scope_string:
+            if not is_client_login():
+                raise click.UsageError(
+                    "Scope requirements (--scope-string) are currently only "
+                    "supported for confidential-client authorized calls."
+                )
+            resource_server = _get_resource_server(service_name)
+            login_manager.add_requirement(resource_server, scope_string)
+
         client = _get_client(login_manager, service_name)
         client.app_name = version.app_name + " raw-api-command"
         if no_retry:

src/globus_cli/login_manager/manager.py

@@ -118,6 +118,13 @@ def has_login(self, resource_server: str) -> bool:
 
         return self._validate_token(tokens["refresh_token"])
 
+    def _tokens_meet_auth_requirements(
+        self, resource_server: str, tokens: dict[str, t.Any]
+    ) -> bool:
+        return self._tokens_meet_static_requirements(
+            resource_server, tokens
+        ) and self._tokens_meet_nonstatic_requirements(resource_server, tokens)
+
     def _tokens_meet_static_requirements(
         self, resource_server: str, tokens: dict[str, t.Any]
     ) -> bool:
@@ -309,7 +316,7 @@ def _get_client_authorizer(
             # or for another client, but automatic retries will handle that
             access_token = None
             expires_at = None
-            if tokens:
+            if tokens and self._tokens_meet_auth_requirements(resource_server, tokens):
                 access_token = tokens["access_token"]
                 expires_at = tokens["expires_at_seconds"]
 

tests/functional/test_api.py

@@ -1,6 +1,7 @@
 import urllib.parse
 
 import pytest
+import responses
 from globus_sdk._testing import (
     RegisteredResponse,
     get_last_request,
@@ -105,3 +106,29 @@ def test_api_command_query_params_multiple_become_list(run_line):
     parsed_query_string = urllib.parse.parse_qs(parsed_url.query)
     assert list(parsed_query_string.keys()) == ["filter"]
     assert set(parsed_query_string["filter"]) == {"frobulated", "demuddled", "reversed"}
+
+
+def test_api_command_with_scope_strings(monkeypatch, client_login, run_line):
+    load_response("cli.api.transfer_stub")
+    load_response("auth.oauth2_client_credentials_tokens")
+
+    run_line("globus api transfer get /foo --scope-string foobarjohn")
+
+    token_grant = [
+        kall for kall in responses.calls if kall.request.url.endswith("/token")
+    ][0]
+    request_params = urllib.parse.parse_qs(token_grant.request.body)
+    assert request_params["grant_type"][0] == "client_credentials"
+    scopes = request_params["scope"][0].split(" ")
+    # This is the default transfer scope, inherited through the service name.
+    assert "urn:globus:auth:scope:transfer.api.globus.org:all" in scopes
+    # This is the scope string we explicitly passed in.
+    assert "foobarjohn" in scopes
+
+
+def test_api_command_rejects_non_client_based_scope_strings(run_line):
+    result = run_line(
+        "globus api auth GET /v2/api/projects --scope-string foobarjohn",
+        assert_exit_code=2,
+    )
+    assert "only supported for confidential-client authorized calls" in result.stderr