Coverage for cli/monitoring_user_mgmt.py: 91.94%

1"""Grafana user management over the admin HTTP API (cluster observability). 

2 

3Grafana uses its own native user database (self sign-up and anonymous access are 

4disabled), so ``gco monitoring users`` drives Grafana's admin HTTP API rather 

5than Cognito. Because Grafana is a private ``ClusterIP`` Service, these calls go 

6through a ``gco monitoring open`` port-forward (default ``http://localhost:3000``). 

7 

8Admin credentials come from the chart-generated ``kube-prometheus-stack-grafana`` 

9Secret (read here via ``kubectl``) unless the caller supplies them explicitly. 

10 

11The HTTP functions are thin, mockable wrappers over ``requests`` (unit-tested by 

12patching ``requests``), mirroring :mod:`cli.analytics_user_mgmt`'s shape. 

13""" 

14 

15from __future__ import annotations 

16 

17import base64 

18import json 

19import re 

20import secrets 

21import subprocess 

22 

23import requests 

24 

25DEFAULT_GRAFANA_URL = "http://localhost:3000" 

26DEFAULT_NAMESPACE = "monitoring" 

27DEFAULT_ADMIN_SECRET = "kube-prometheus-stack-grafana" 

28 

29_HTTP_TIMEOUT_SECONDS = 15 

30_NAMESPACE_RE = re.compile(r"^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?$") 

31_SECRET_RE = re.compile(r"^[a-z0-9]([a-z0-9.\-]{0,251}[a-z0-9])?$") 

32 

33 

34def generate_password(n_bytes: int = 24) -> str: 

35 """Return a cryptographically strong, URL-safe password.""" 

36 return secrets.token_urlsafe(n_bytes) 

37 

38 

39def read_grafana_admin_credentials( 

40 namespace: str = DEFAULT_NAMESPACE, secret_name: str = DEFAULT_ADMIN_SECRET 

41) -> tuple[str, str]: 

42 """Read ``(admin_user, admin_password)`` from the Grafana Secret via kubectl. 

43 

44 Requires cluster access (the same connectivity ``gco monitoring open`` needs). 

45 Raises ``RuntimeError`` if kubectl fails or the keys are absent. 

46 """ 

47 if not _NAMESPACE_RE.match(namespace): 

48 raise ValueError(f"Invalid namespace {namespace!r}") 

49 if not _SECRET_RE.match(secret_name): 49 ↛ 50line 49 didn't jump to line 50 because the condition on line 49 was never true

50 raise ValueError(f"Invalid secret name {secret_name!r}") 

51 

52 cmd = ["kubectl", "get", "secret", secret_name, "-n", namespace, "-o", "json"] 

53 try: 

54 result = subprocess.run( 

55 cmd, capture_output=True, text=True 

56 ) # nosemgrep: dangerous-subprocess-use-audit - inputs validated above; list form, no shell=True 

57 except FileNotFoundError as exc: 

58 raise RuntimeError( 

59 "kubectl not found. Install kubectl and ensure it's on your PATH." 

60 ) from exc 

61 if result.returncode != 0: 

62 raise RuntimeError( 

63 f"Failed to read Secret {namespace}/{secret_name}: {result.stderr.strip()}" 

64 ) 

65 

66 data = json.loads(result.stdout or "{}").get("data", {}) 

67 if "admin-user" not in data or "admin-password" not in data: 

68 raise RuntimeError( 

69 f"Secret {namespace}/{secret_name} does not carry admin-user/admin-password" 

70 ) 

71 user = base64.b64decode(data["admin-user"]).decode("utf-8") 

72 password = base64.b64decode(data["admin-password"]).decode("utf-8") 

73 return user, password 

74 

75 

76def create_user( 

77 base_url: str, 

78 auth: tuple[str, str], 

79 *, 

80 login: str, 

81 password: str, 

82 email: str | None = None, 

83 name: str | None = None, 

84) -> int: 

85 """Create a Grafana user via ``POST /api/admin/users``; return the new id.""" 

86 body: dict[str, str] = {"login": login, "password": password, "name": name or login} 

87 if email: 87 ↛ 89line 87 didn't jump to line 89 because the condition on line 87 was always true

88 body["email"] = email 

89 resp = requests.post( 

90 f"{base_url}/api/admin/users", auth=auth, json=body, timeout=_HTTP_TIMEOUT_SECONDS 

91 ) 

92 resp.raise_for_status() 

93 return int(resp.json()["id"]) 

94 

95 

96def list_users(base_url: str, auth: tuple[str, str]) -> list[dict[str, object]]: 

97 """List organisation users via ``GET /api/org/users``.""" 

98 resp = requests.get(f"{base_url}/api/org/users", auth=auth, timeout=_HTTP_TIMEOUT_SECONDS) 

99 resp.raise_for_status() 

100 result = resp.json() 

101 return result if isinstance(result, list) else [] 

102 

103 

104def lookup_user_id(base_url: str, auth: tuple[str, str], login_or_email: str) -> int: 

105 """Resolve a login/email to a numeric user id via ``GET /api/users/lookup``.""" 

106 resp = requests.get( 

107 f"{base_url}/api/users/lookup", 

108 params={"loginOrEmail": login_or_email}, 

109 auth=auth, 

110 timeout=_HTTP_TIMEOUT_SECONDS, 

111 ) 

112 resp.raise_for_status() 

113 return int(resp.json()["id"]) 

114 

115 

116def delete_user(base_url: str, auth: tuple[str, str], user_id: int) -> None: 

117 """Delete a Grafana user via ``DELETE /api/admin/users/<id>``.""" 

118 resp = requests.delete( 

119 f"{base_url}/api/admin/users/{int(user_id)}", 

120 auth=auth, 

121 timeout=_HTTP_TIMEOUT_SECONDS, 

122 ) 

123 resp.raise_for_status()