Coverage for gco/services/grafana_rotator.py: 96.83%

1"""Scheduled rotation of the Grafana admin password (cluster observability). 

2 

3Runs as an in-cluster ``CronJob`` (see 

4``lambda/kubectl-applier-simple/manifests/post-helm-grafana-credential-rotation.yaml``) 

5using an existing GCO service image, so no extra container image is introduced. 

6 

7Why an active reset rather than a restart: the kube-prometheus-stack Grafana 

8subchart auto-generates the admin credential into the ``<release>-grafana`` 

9Secret and Grafana persists it in its own database. ``GF_SECURITY_ADMIN_PASSWORD`` 

10only seeds the password on Grafana's *first* start, so patching the Secret and 

11restarting the pod does not change the live password. Rotation therefore resets 

12the password through Grafana's admin HTTP API and then updates the Secret that 

13``gco monitoring`` reads at runtime. 

14 

15Order is admin-API-first so the live password and the Secret only diverge for 

16the single Secret-patch call that follows a successful reset: 

17 

18 1. Read the current admin user/password from the Secret. 

19 2. Generate a new strong password. 

20 3. ``GET /api/user`` -> the admin user's numeric id. 

21 4. ``PUT /api/admin/users/<id>/password`` with the new password. 

22 5. Patch the Secret's ``admin-password`` to the new value. 

23 

24Least privilege: the only Kubernetes permission the CronJob's ServiceAccount is 

25granted is ``get``/``patch`` on the single Grafana Secret in the ``monitoring`` 

26namespace. The reset goes through the ClusterIP Service, so no ``pods/exec`` and 

27no AWS/IRSA permissions are required. 

28""" 

29 

30import base64 

31import logging 

32import os 

33import secrets 

34import sys 

35 

36import requests 

37from kubernetes import client, config 

38from kubernetes.client.rest import ApiException 

39 

40logger = logging.getLogger("gco.grafana_rotator") 

41 

42# Defaults match the kube-prometheus-stack release name GCO installs 

43# (the helm-installer uses the chart key as the release name, so the Grafana 

44# subchart resources are named ``kube-prometheus-stack-grafana``). All three are 

45# overridable via env so the manifest stays the single source of truth. 

46DEFAULT_NAMESPACE = "monitoring" 

47DEFAULT_SECRET_NAME = "kube-prometheus-stack-grafana" 

48DEFAULT_SERVICE_URL = "http://kube-prometheus-stack-grafana.monitoring.svc" 

49 

50_ADMIN_USER_KEY = "admin-user" 

51_ADMIN_PASSWORD_KEY = "admin-password" 

52 

53# 24 random bytes -> ~32 URL-safe characters. Comfortably above any sane 

54# minimum-length policy while staying a valid Grafana password. 

55_PASSWORD_ENTROPY_BYTES = 24 

56_HTTP_TIMEOUT_SECONDS = 15 

57 

58 

59def generate_password(n_bytes: int = _PASSWORD_ENTROPY_BYTES) -> str: 

60 """Return a cryptographically strong, URL-safe password string.""" 

61 return secrets.token_urlsafe(n_bytes) 

62 

63 

64def read_admin_credentials( 

65 core_v1: client.CoreV1Api, namespace: str, secret_name: str 

66) -> tuple[str, str]: 

67 """Return ``(admin_user, admin_password)`` decoded from the Grafana Secret.""" 

68 secret = core_v1.read_namespaced_secret(secret_name, namespace) 

69 data = secret.data or {} 

70 missing = [k for k in (_ADMIN_USER_KEY, _ADMIN_PASSWORD_KEY) if k not in data] 

71 if missing: 

72 raise KeyError(f"Secret {namespace}/{secret_name} is missing key(s): {', '.join(missing)}") 

73 user = base64.b64decode(data[_ADMIN_USER_KEY]).decode("utf-8") 

74 password = base64.b64decode(data[_ADMIN_PASSWORD_KEY]).decode("utf-8") 

75 return user, password 

76 

77 

78def get_admin_user_id(service_url: str, auth: tuple[str, str]) -> int: 

79 """Look up the authenticated admin user's numeric id via ``GET /api/user``. 

80 

81 Resolving the id at runtime avoids hardcoding the bootstrap admin id (1), 

82 which would silently target the wrong account if the org admin ever changes. 

83 """ 

84 resp = requests.get(f"{service_url}/api/user", auth=auth, timeout=_HTTP_TIMEOUT_SECONDS) 

85 resp.raise_for_status() 

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

87 

88 

89def reset_admin_password( 

90 service_url: str, auth: tuple[str, str], user_id: int, new_password: str 

91) -> None: 

92 """Reset the admin password through Grafana's admin API.""" 

93 resp = requests.put( 

94 f"{service_url}/api/admin/users/{user_id}/password", 

95 auth=auth, 

96 json={"password": new_password}, 

97 timeout=_HTTP_TIMEOUT_SECONDS, 

98 ) 

99 resp.raise_for_status() 

100 

101 

102def patch_secret_password( 

103 core_v1: client.CoreV1Api, namespace: str, secret_name: str, new_password: str 

104) -> None: 

105 """Write the new password back into the Grafana Secret (base64 ``data``).""" 

106 encoded = base64.b64encode(new_password.encode("utf-8")).decode("utf-8") 

107 core_v1.patch_namespaced_secret( 

108 secret_name, namespace, {"data": {_ADMIN_PASSWORD_KEY: encoded}} 

109 ) 

110 

111 

112def rotate(core_v1: client.CoreV1Api, namespace: str, secret_name: str, service_url: str) -> None: 

113 """Rotate the Grafana admin password end to end. 

114 

115 Reads the current credential, resets the live password through the admin 

116 API, then persists the new value to the Secret. The password itself is 

117 never logged. 

118 """ 

119 user, current_password = read_admin_credentials(core_v1, namespace, secret_name) 

120 new_password = generate_password() 

121 auth = (user, current_password) 

122 user_id = get_admin_user_id(service_url, auth) 

123 reset_admin_password(service_url, auth, user_id, new_password) 

124 patch_secret_password(core_v1, namespace, secret_name, new_password) 

125 # Audit line carries only non-secret identifiers (username, user id, service 

126 # URL, namespace/secret names). The password itself is never logged. 

127 # nosemgrep: python-logger-credential-disclosure 

128 logger.info( 

129 "Rotated Grafana admin credential for user %r (id=%s) via %s (secret %s/%s)", 

130 user, 

131 user_id, 

132 service_url, 

133 namespace, 

134 secret_name, 

135 ) 

136 

137 

138def main() -> int: 

139 """CronJob entrypoint. Returns a process exit code (0 ok, 1 on failure).""" 

140 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") 

141 namespace = os.environ.get("GRAFANA_NAMESPACE", DEFAULT_NAMESPACE) 

142 secret_name = os.environ.get("GRAFANA_ADMIN_SECRET", DEFAULT_SECRET_NAME) 

143 service_url = os.environ.get("GRAFANA_SERVICE_URL", DEFAULT_SERVICE_URL).rstrip("/") 

144 

145 try: 

146 config.load_incluster_config() 

147 except config.ConfigException: 

148 # Allows local dry-runs against a configured kubeconfig; in the CronJob 

149 # the in-cluster path always wins. 

150 config.load_kube_config() 

151 

152 core_v1 = client.CoreV1Api() 

153 try: 

154 rotate(core_v1, namespace, secret_name, service_url) 

155 except (ApiException, requests.RequestException, KeyError, ValueError) as exc: 

156 # Only the caught exception is logged; it carries no credential material. 

157 # nosemgrep: python-logger-credential-disclosure 

158 logger.error("Grafana admin credential rotation failed: %s", exc) 

159 return 1 

160 return 0 

161 

162 

163if __name__ == "__main__": 

164 sys.exit(main())