Coverage for cli / monitoring_user_mgmt.py: 100.00%

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" # nosec B105 # Kubernetes Secret name, not its contents 

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): 

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, 

56 capture_output=True, 

57 text=True, 

58 check=False, 

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

60 except FileNotFoundError as exc: 

61 raise RuntimeError( 

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

63 ) from exc 

64 if result.returncode != 0: 

65 raise RuntimeError( 

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

67 ) 

68 

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

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

71 raise RuntimeError( 

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

73 ) 

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

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

76 return user, password 

77 

78 

79def create_user( 

80 base_url: str, 

81 auth: tuple[str, str], 

82 *, 

83 login: str, 

84 password: str, 

85 email: str | None = None, 

86 name: str | None = None, 

87) -> int: 

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

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

90 if email: 

91 body["email"] = email 

92 resp = requests.post( 

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

94 ) 

95 resp.raise_for_status() 

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

97 

98 

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

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

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

102 resp.raise_for_status() 

103 result = resp.json() 

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

105 

106 

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

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

109 resp = requests.get( 

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

111 params={"loginOrEmail": login_or_email}, 

112 auth=auth, 

113 timeout=_HTTP_TIMEOUT_SECONDS, 

114 ) 

115 resp.raise_for_status() 

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

117 

118 

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

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

121 resp = requests.delete( 

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

123 auth=auth, 

124 timeout=_HTTP_TIMEOUT_SECONDS, 

125 ) 

126 resp.raise_for_status()