Coverage for cli/analytics_user_mgmt.py: 92.92%
184 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""
2User management helpers for the GCO analytics environment.
4This module holds the pieces of the ``gco analytics`` CLI that are worth
5exercising in isolation from Click:
7* :func:`discover_cognito_pool_id` / :func:`discover_cognito_client_id`
8 / :func:`discover_api_endpoint` — single-stack CloudFormation output
9 lookups used by every sub-command to avoid forcing operators to hand a
10 pool id / api url on the command line.
11* :func:`srp_authenticate` — Cognito SRP authentication via the
12 ``pycognito`` library, used by ``gco analytics studio login``.
13"""
15from __future__ import annotations
17import logging
18from typing import Any
20# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
21# Generated at (UTC): 2026-07-18T01:03:40Z
22# Flowchart(s) generated from this file:
23# * ``srp_authenticate`` -> ``diagrams/code_diagrams/cli/analytics_user_mgmt.srp_authenticate.html``
24# (PNG: ``diagrams/code_diagrams/cli/analytics_user_mgmt.srp_authenticate.png``)
25# * ``fetch_studio_url`` -> ``diagrams/code_diagrams/cli/analytics_user_mgmt.fetch_studio_url.html``
26# (PNG: ``diagrams/code_diagrams/cli/analytics_user_mgmt.fetch_studio_url.png``)
27# Regenerate with ``python diagrams/code_diagrams/generate.py``.
28# <pyflowchart-code-diagram> END
31logger = logging.getLogger(__name__)
33# ---------------------------------------------------------------------------
34# CloudFormation output discovery
35# ---------------------------------------------------------------------------
38def _describe_stack_outputs(region: str, stack_name: str) -> list[dict[str, str]] | None:
39 """Return the ``Outputs`` list for ``stack_name`` in ``region``.
41 Returns ``None`` if the stack does not exist or the call fails.
42 Any non-transient error surfaces as ``None`` — callers raise the
43 user-facing error message themselves so the error copy can mention
44 ``gco analytics enable`` / ``gco stacks deploy gco-analytics``.
45 """
46 import boto3
47 from botocore.exceptions import BotoCoreError, ClientError
49 try:
50 cfn = boto3.client("cloudformation", region_name=region)
51 response = cfn.describe_stacks(StackName=stack_name)
52 except (ClientError, BotoCoreError) as exc:
53 logger.debug("describe_stacks(%s) in %s failed: %s", stack_name, region, exc)
54 return None
56 stacks = response.get("Stacks", [])
57 if not stacks: 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true
58 return None
59 outputs = stacks[0].get("Outputs", [])
60 return list(outputs) if isinstance(outputs, list) else []
63def _find_output(outputs: list[dict[str, str]], key: str) -> str | None:
64 """Return the ``OutputValue`` for ``key`` in a CloudFormation outputs list."""
65 for output in outputs:
66 if output.get("OutputKey") == key:
67 value = output.get("OutputValue")
68 return value if isinstance(value, str) else None
69 return None
72def discover_cognito_pool_id(region: str, project_name: str = "gco") -> str | None:
73 """Return the Cognito user pool id published by ``gco-analytics``.
75 Returns ``None`` when the ``gco-analytics`` stack does not exist or
76 when the stack exists but the ``CognitoUserPoolId`` output is
77 missing. The CLI callers translate ``None`` into the documented
78 "gco-analytics stack not deployed" error message.
79 """
80 stack_name = f"{project_name}-analytics"
81 outputs = _describe_stack_outputs(region, stack_name)
82 if outputs is None:
83 return None
84 return _find_output(outputs, "CognitoUserPoolId")
87def discover_cognito_client_id(region: str, project_name: str = "gco") -> str | None:
88 """Return the Cognito SRP client id published by ``gco-analytics``.
90 Looked up on the same stack as :func:`discover_cognito_pool_id`.
91 Returns ``None`` when the stack or output is missing.
92 """
93 stack_name = f"{project_name}-analytics"
94 outputs = _describe_stack_outputs(region, stack_name)
95 if outputs is None:
96 return None
97 return _find_output(outputs, "CognitoUserPoolClientId")
100def discover_api_endpoint(region: str, project_name: str = "gco") -> str | None:
101 """Return the API Gateway base URL published by ``gco-api-gateway``.
103 The returned value is the ``ApiEndpoint`` CloudFormation output,
104 typically of the form ``https://<id>.execute-api.<region>.amazonaws.com/prod/``.
105 Returns ``None`` when the stack or output is missing.
106 """
107 stack_name = f"{project_name}-api-gateway"
108 outputs = _describe_stack_outputs(region, stack_name)
109 if outputs is None:
110 return None
111 return _find_output(outputs, "ApiEndpoint")
114# ---------------------------------------------------------------------------
115# Cognito authentication
116# ---------------------------------------------------------------------------
119def srp_authenticate(
120 pool_id: str,
121 client_id: str,
122 username: str,
123 password: str,
124 region: str,
125) -> dict[str, str]:
126 """Authenticate a Cognito user via the ADMIN_USER_PASSWORD_AUTH flow.
128 Uses ``admin_initiate_auth`` which sends the password over TLS
129 directly (no client-side SRP math). This requires the user pool
130 client to have ``ALLOW_ADMIN_USER_PASSWORD_AUTH`` enabled and the
131 caller to have ``cognito-idp:AdminInitiateAuth`` permission.
133 Returns a dict with ``IdToken``, ``AccessToken``, and
134 ``RefreshToken`` on success. Raises ``botocore.exceptions.ClientError``
135 for Cognito-side failures (``NotAuthorizedException``,
136 ``UserNotFoundException``, etc.).
137 """
138 import boto3
140 cognito = boto3.client("cognito-idp", region_name=region)
141 response = cognito.admin_initiate_auth(
142 UserPoolId=pool_id,
143 ClientId=client_id,
144 AuthFlow="ADMIN_USER_PASSWORD_AUTH",
145 AuthParameters={
146 "USERNAME": username,
147 "PASSWORD": password,
148 },
149 )
150 tokens = response.get("AuthenticationResult") or {}
151 return {
152 "IdToken": str(tokens.get("IdToken", "")),
153 "AccessToken": str(tokens.get("AccessToken", "")),
154 "RefreshToken": str(tokens.get("RefreshToken", "")),
155 }
158__all__ = [
159 "admin_create_user",
160 "admin_delete_user",
161 "admin_set_user_password",
162 "check_ssm_parameter",
163 "check_stack_complete",
164 "discover_api_endpoint",
165 "discover_cognito_client_id",
166 "discover_cognito_pool_id",
167 "fetch_studio_url",
168 "generate_strong_password",
169 "list_users",
170 "scan_orphan_analytics_resources",
171 "srp_authenticate",
172]
174# ---------------------------------------------------------------------------
175# Cognito user management helpers
176# ---------------------------------------------------------------------------
179def admin_create_user(
180 pool_id: str,
181 region: str,
182 username: str,
183 email: str | None = None,
184 suppress_email: bool = False,
185) -> tuple[dict[str, Any], str | None]:
186 """Create a Cognito user via AdminCreateUser.
188 Returns ``(response, temporary_password)``. The temporary password
189 is only set when Cognito echoes it in the response (it does this
190 on some versions of the API when ``MessageAction=SUPPRESS``); when
191 absent the caller should direct the operator to
192 ``admin-set-user-password`` out-of-band.
193 """
194 import boto3
196 user_attributes: list[dict[str, str]] = []
197 if email:
198 user_attributes.append({"Name": "email", "Value": email})
199 user_attributes.append({"Name": "email_verified", "Value": "true"})
201 kwargs: dict[str, Any] = {
202 "UserPoolId": pool_id,
203 "Username": username,
204 "UserAttributes": user_attributes,
205 }
206 if suppress_email:
207 kwargs["MessageAction"] = "SUPPRESS"
209 cognito = boto3.client("cognito-idp", region_name=region)
210 response = cognito.admin_create_user(**kwargs)
212 temporary_password: str | None = None
213 user = response.get("User", {})
214 for attr in user.get("Attributes", []) or []:
215 if attr.get("Name") == "temporary_password":
216 temporary_password = attr.get("Value")
217 break
218 if temporary_password is None:
219 temporary_password = response.get("TemporaryPassword")
221 return response, temporary_password
224def admin_set_user_password(
225 pool_id: str,
226 region: str,
227 username: str,
228 password: str,
229 permanent: bool = True,
230) -> None:
231 """Set a Cognito user's password via AdminSetUserPassword.
233 ``permanent=True`` (the default) marks the password as already
234 satisfying Cognito's ``NEW_PASSWORD_REQUIRED`` challenge so the
235 user can sign in without a forced reset — matching what you'd
236 get from ``aws cognito-idp admin-set-user-password --permanent``.
237 Pass ``permanent=False`` to require the user to pick their own
238 password on first login.
239 """
240 import boto3
242 cognito = boto3.client("cognito-idp", region_name=region)
243 cognito.admin_set_user_password(
244 UserPoolId=pool_id,
245 Username=username,
246 Password=password,
247 Permanent=permanent,
248 )
251def generate_strong_password(length: int = 20) -> str:
252 """Return a random password that satisfies Cognito's default policy.
254 Cognito's default password policy requires at least one uppercase
255 letter, one lowercase letter, one digit, and one symbol, plus the
256 length minimum (8). The generated password is sampled from
257 :func:`secrets.choice` — cryptographically strong by construction —
258 and guaranteed to contain one character from each required class,
259 with the remaining characters drawn from the union.
260 """
261 import secrets
262 import string
264 if length < 8:
265 raise ValueError(f"length must be >= 8 to satisfy Cognito policy; got {length}")
267 lowers = string.ascii_lowercase
268 uppers = string.ascii_uppercase
269 digits = string.digits
270 # Cognito's allowed symbol set per AWS docs. Notably excludes space
271 # and tab — Cognito rejects whitespace with InvalidParameterException.
272 symbols = "^$*.[]{}()?-\"!@#%&/\\,><':;|_~`+="
274 required = [
275 secrets.choice(lowers),
276 secrets.choice(uppers),
277 secrets.choice(digits),
278 secrets.choice(symbols),
279 ]
280 alphabet = lowers + uppers + digits + symbols
281 remaining = [secrets.choice(alphabet) for _ in range(length - len(required))]
283 # Shuffle so the required-class characters aren't always at the start.
284 chars = required + remaining
285 for i in range(len(chars) - 1, 0, -1):
286 j = secrets.randbelow(i + 1)
287 chars[i], chars[j] = chars[j], chars[i]
289 return "".join(chars)
292def list_users(pool_id: str, region: str) -> list[dict[str, str]]:
293 """Return a flat row-per-user list suitable for tabular output."""
294 import boto3
296 cognito = boto3.client("cognito-idp", region_name=region)
297 response = cognito.list_users(UserPoolId=pool_id)
299 rows: list[dict[str, str]] = []
300 for user in response.get("Users", []) or []:
301 row: dict[str, str] = {
302 "username": user.get("Username", ""),
303 "status": user.get("UserStatus", ""),
304 "enabled": str(user.get("Enabled", "")),
305 }
306 for attr in user.get("Attributes", []) or []:
307 if attr.get("Name") == "email":
308 row["email"] = attr.get("Value", "")
309 rows.append(row)
310 return rows
313def admin_delete_user(pool_id: str, region: str, username: str) -> None:
314 """Delete a Cognito user via AdminDeleteUser."""
315 import boto3
317 cognito = boto3.client("cognito-idp", region_name=region)
318 cognito.admin_delete_user(UserPoolId=pool_id, Username=username)
321# ---------------------------------------------------------------------------
322# HTTP helper for /studio/login
323# ---------------------------------------------------------------------------
326def fetch_studio_url(api_base: str, id_token: str) -> tuple[str, int, str]:
327 """GET ``{api_base}/studio/login`` with the Cognito ID token.
329 Returns ``(url, expires_in, correlation_id)`` on success. Raises
330 :class:`urllib.error.HTTPError` / :class:`urllib.error.URLError`
331 on transport or HTTP failure; raises ``ValueError`` on malformed
332 response bodies (unexpected JSON shape / missing ``url`` key), or
333 on non-``https://`` ``api_base`` values (guards urllib's
334 ``file://`` / ``ftp://`` scheme support).
335 """
336 import email.message
337 import json as _json
338 import urllib.error
339 import urllib.parse
340 import urllib.request
342 # Scheme allow-list — urllib happily dereferences ``file://`` and
343 # ``ftp://`` URLs, which is the shape of the semgrep
344 # ``dynamic-urllib-use-detected`` finding. We only ever call this with
345 # the API Gateway endpoint (HTTPS by construction), so reject anything
346 # else before the urlopen call.
347 parsed = urllib.parse.urlparse(api_base)
348 if parsed.scheme != "https": 348 ↛ 349line 348 didn't jump to line 349 because the condition on line 348 was never true
349 raise ValueError(
350 f"api_base must use https:// scheme (got {parsed.scheme!r}). "
351 "This guard rejects file:// / ftp:// schemes that urllib would "
352 "otherwise follow."
353 )
354 if not parsed.netloc: 354 ↛ 355line 354 didn't jump to line 355 because the condition on line 354 was never true
355 raise ValueError(f"api_base is missing a hostname: {api_base!r}")
357 login_url = api_base.rstrip("/") + "/studio/login"
358 # Justification for the ``dynamic-urllib-use-detected`` / ``B310``
359 # suppressions below: ``login_url`` is built from ``api_base`` + a
360 # static ``/studio/login`` suffix. The scheme allow-list near the top
361 # of this function rejects any ``api_base`` that isn't ``https://``
362 # before we reach these lines, which closes the ``file://`` /
363 # ``ftp://`` / ``custom`` scheme hole the rules are written to catch.
364 # ``# fmt: off`` pins the block so the formatter can't re-wrap the
365 # urlopen call — wrapping moves the suppression comments to the
366 # wrong line and bandit / semgrep attach findings to the first
367 # line of the call.
368 # fmt: off
369 request = urllib.request.Request( # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected # noqa: S310
370 login_url,
371 headers={"Authorization": id_token, "Accept": "application/json"},
372 method="GET",
373 )
374 with urllib.request.urlopen(request, timeout=30) as response: # nosec B310 # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected # noqa: S310
375 status = int(response.status)
376 body = response.read().decode("utf-8")
377 correlation_id = response.headers.get("x-amzn-RequestId") or "N/A"
378 # fmt: on
380 if status == 202: 380 ↛ 383line 380 didn't jump to line 383 because the condition on line 380 was never true
381 # Profile is still provisioning -- return empty URL so the caller
382 # can poll. The body is ``{"status": "provisioning"}``.
383 return "", 0, correlation_id
385 if status != 200:
386 # HTTPError requires a Message (email.message.Message) as its
387 # headers argument; build an empty one for determinism.
388 headers_msg: email.message.Message = email.message.Message()
389 headers_msg["x-amzn-RequestId"] = correlation_id
390 raise urllib.error.HTTPError(
391 login_url,
392 status,
393 f"Studio login returned HTTP {status}",
394 headers_msg,
395 None,
396 )
398 try:
399 payload = _json.loads(body)
400 url = str(payload["url"])
401 expires_in = int(payload.get("expires_in", 0))
402 except (ValueError, KeyError) as exc:
403 raise ValueError(f"malformed /studio/login response: {exc!r}") from exc
405 return url, expires_in, correlation_id
408# ---------------------------------------------------------------------------
409# Doctor helpers
410# ---------------------------------------------------------------------------
413def check_stack_complete(region: str, stack_name: str) -> tuple[bool, str]:
414 """Return ``(True, "")`` iff ``stack_name`` is in a healthy state.
416 Healthy states are ``CREATE_COMPLETE`` / ``UPDATE_COMPLETE`` /
417 ``IMPORT_COMPLETE``. Any other status (or missing stack) returns
418 ``(False, remediation_hint)``.
419 """
420 import boto3
421 from botocore.exceptions import BotoCoreError, ClientError
423 try:
424 cfn = boto3.client("cloudformation", region_name=region)
425 resp = cfn.describe_stacks(StackName=stack_name)
426 except (ClientError, BotoCoreError) as exc:
427 return False, f"describe_stacks failed in {region}: {exc!s}"
428 stacks = resp.get("Stacks", [])
429 if not stacks:
430 return False, f"{stack_name} not found in {region}"
431 status = stacks[0].get("StackStatus", "")
432 if status in ("CREATE_COMPLETE", "UPDATE_COMPLETE", "IMPORT_COMPLETE"):
433 return True, ""
434 return False, f"{stack_name} in {region} has status {status}"
437def check_ssm_parameter(region: str, param_name: str) -> tuple[bool, str]:
438 """Return ``(True, "")`` iff the SSM parameter exists in ``region``.
440 Thin alias over :func:`gco.services.aws_ssm.check_ssm_parameter`
441 that preserves the historical positional ``(region, param_name)``
442 argument order. Kept as a re-export so existing callers and the
443 public ``__all__`` surface stay stable; new code should reach for
444 the keyword-style helper directly.
445 """
446 from gco.services.aws_ssm import check_ssm_parameter as _check
448 return _check(param_name, region=region)
451def scan_orphan_analytics_resources(region: str) -> list[str]:
452 """Return a list of copy-paste ``aws`` commands for retained resources.
454 Scans EFS and Cognito for resources tagged
455 ``gco:analytics:managed=true``. An empty list means no orphans
456 were found.
457 """
458 import boto3
459 from botocore.exceptions import BotoCoreError, ClientError
461 remediation: list[str] = []
462 try:
463 efs = boto3.client("efs", region_name=region)
464 for fs in efs.describe_file_systems().get("FileSystems", []) or []:
465 fs_id = fs.get("FileSystemId", "")
466 if not fs_id: 466 ↛ 467line 466 didn't jump to line 467 because the condition on line 466 was never true
467 continue
468 tag_resp = efs.list_tags_for_resource(ResourceId=fs_id)
469 tags = {t.get("Key"): t.get("Value") for t in tag_resp.get("Tags", []) or []}
470 if tags.get("gco:analytics:managed") == "true":
471 remediation.append(f"aws efs delete-file-system --file-system-id {fs_id}")
472 except (ClientError, BotoCoreError) as exc:
473 remediation.append(f"(EFS orphan scan failed: {exc!s})")
475 try:
476 cognito = boto3.client("cognito-idp", region_name=region)
477 pools = cognito.list_user_pools(MaxResults=60)
478 for pool in pools.get("UserPools", []) or []:
479 pool_id = pool.get("Id")
480 if not pool_id: 480 ↛ 481line 480 didn't jump to line 481 because the condition on line 480 was never true
481 continue
482 describe = cognito.describe_user_pool(UserPoolId=pool_id)
483 tags = describe.get("UserPool", {}).get("UserPoolTags", {}) or {}
484 if tags.get("gco:analytics:managed") == "true":
485 remediation.append(f"aws cognito-idp delete-user-pool --user-pool-id {pool_id}")
486 except (ClientError, BotoCoreError) as exc:
487 remediation.append(f"(Cognito orphan scan failed: {exc!s})")
489 return remediation