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

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 a ClusterIP Service, so no ``pods/exec`` and 

27no AWS/IRSA permissions are required. 

28 

29Transport: the admin API calls carry the admin credential, so they go over 

30verified HTTPS to the ``grafana-tls`` Service, whose ``grafana-tls-proxy`` 

31sidecar terminates TLS with a leaf issued by the GCO internal CA. The rotator 

32trusts that CA alone (the ``ca.crt`` key projected at ``/var/run/gco/ca``, or 

33``GCO_INTERNAL_CA_FILE``) and verifies the Service host name. An ``https`` URL 

34with no usable CA bundle fails closed before any credential is read; a plain 

35``http`` override (local runs against a port-forward) opens no TLS session and 

36needs no bundle. 

37""" 

38 

39from __future__ import annotations 

40 

41import base64 

42import logging 

43import os 

44import secrets 

45import sys 

46from urllib.parse import urlsplit 

47 

48import requests 

49from kubernetes import client, config 

50from kubernetes.client.rest import ApiException 

51 

52from gco.services import internal_tls 

53 

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

55 

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

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

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

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

60DEFAULT_NAMESPACE = "monitoring" 

61DEFAULT_SECRET_NAME = "kube-prometheus-stack-grafana" # nosec B105 # Kubernetes Secret name, not its contents 

62# The TLS front door GCO adds beside the chart's plaintext Service 

63# (post-helm-monitoring-tls.yaml): 3443 -> the grafana-tls-proxy sidecar -> 

64# Grafana over the pod's loopback. 

65DEFAULT_SERVICE_URL = "https://grafana-tls.monitoring.svc.cluster.local:3443" 

66 

67_ADMIN_USER_KEY = "admin-user" 

68_ADMIN_PASSWORD_KEY = "admin-password" # nosec B105 # data key inside the Secret, not the value 

69 

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

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

72_PASSWORD_ENTROPY_BYTES = 24 

73_HTTP_TIMEOUT_SECONDS = 15 

74 

75 

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

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

78 return secrets.token_urlsafe(n_bytes) 

79 

80 

81def tls_verify_for(service_url: str) -> str | bool: 

82 """Return the ``requests`` ``verify=`` value for the Grafana admin URL. 

83 

84 ``https`` verifies against the GCO internal CA bundle only: ``requests`` 

85 given a bundle path loads no other trust anchors, and host name checking 

86 stays on. The bundle is parsed here first, so a missing or unusable file 

87 raises :class:`~gco.services.internal_tls.InternalTLSError` naming the path 

88 and the variable to set, instead of failing later inside ``requests`` with 

89 the admin credential already in hand. Any other scheme opens no TLS 

90 session, so the library default is returned and no bundle is required. 

91 """ 

92 if urlsplit(service_url).scheme.lower() != "https": 

93 return True 

94 ca_file = internal_tls.internal_ca_file() 

95 try: 

96 internal_tls.internal_ssl_context(ca_file) 

97 except internal_tls.InternalTLSError as exc: 

98 raise internal_tls.InternalTLSError( 

99 f"refusing to call {service_url} without the GCO internal CA: {exc}; mount the " 

100 f"ca.crt key at {ca_file} or point {internal_tls.INTERNAL_CA_FILE_ENV} at it" 

101 ) from exc 

102 return str(ca_file) 

103 

104 

105def read_admin_credentials( 

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

107) -> tuple[str, str]: 

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

109 secret = core_v1.read_namespaced_secret(secret_name, namespace) 

110 data = secret.data or {} 

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

112 if missing: 

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

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

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

116 return user, password 

117 

118 

119def get_admin_user_id(service_url: str, auth: tuple[str, str], verify: str | bool = True) -> int: 

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

121 

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

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

124 """ 

125 resp = requests.get( 

126 f"{service_url}/api/user", auth=auth, timeout=_HTTP_TIMEOUT_SECONDS, verify=verify 

127 ) 

128 resp.raise_for_status() 

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

130 

131 

132def reset_admin_password( 

133 service_url: str, 

134 auth: tuple[str, str], 

135 user_id: int, 

136 new_password: str, 

137 verify: str | bool = True, 

138) -> None: 

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

140 resp = requests.put( 

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

142 auth=auth, 

143 json={"password": new_password}, 

144 timeout=_HTTP_TIMEOUT_SECONDS, 

145 verify=verify, 

146 ) 

147 resp.raise_for_status() 

148 

149 

150def patch_secret_password( 

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

152) -> None: 

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

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

155 core_v1.patch_namespaced_secret( 

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

157 ) 

158 

159 

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

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

162 

163 Resolves the transport trust first (so a missing CA bundle fails before 

164 any credential is read), reads the current credential, resets the live 

165 password through the admin API, then persists the new value to the 

166 Secret. The password itself is never logged. 

167 """ 

168 verify = tls_verify_for(service_url) 

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

170 new_password = generate_password() 

171 auth = (user, current_password) 

172 user_id = get_admin_user_id(service_url, auth, verify) 

173 reset_admin_password(service_url, auth, user_id, new_password, verify) 

174 patch_secret_password(core_v1, namespace, secret_name, new_password) 

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

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

177 # nosemgrep: python-logger-credential-disclosure 

178 logger.info( 

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

180 user, 

181 user_id, 

182 service_url, 

183 namespace, 

184 secret_name, 

185 ) 

186 

187 

188def main() -> int: 

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

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

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

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

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

194 

195 try: 

196 config.load_incluster_config() 

197 except config.ConfigException: 

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

199 # the in-cluster path always wins. 

200 config.load_kube_config() 

201 

202 core_v1 = client.CoreV1Api() 

203 try: 

204 rotate(core_v1, namespace, secret_name, service_url) 

205 except ( 

206 ApiException, 

207 requests.RequestException, 

208 internal_tls.InternalTLSError, 

209 KeyError, 

210 ValueError, 

211 ) as exc: 

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

213 # nosemgrep: python-logger-credential-disclosure 

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

215 return 1 

216 return 0 

217 

218 

219if __name__ == "__main__": 

220 sys.exit(main())