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