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

1"""Trust configuration for GCO's in-cluster HTTPS clients. 

2 

3Every in-cluster hop GCO owns terminates TLS with a leaf certificate that 

4cert-manager issues from the ``gco-internal-ca`` ClusterIssuer. A client of 

5one of those hops verifies the peer against that CA alone, never the public 

6trust store: a certificate minted by any public CA is rejected, and so is a 

7peer whose certificate does not name the Service host the client dialled. 

8 

9The CA reaches a client as the ``ca.crt`` key of a cert-manager Secret in the 

10client's own namespace. cert-manager writes the issuing CA's certificate into 

11every leaf Secret it signs, so a pod that already mounts its own serving 

12keypair can project just that one key, on its own, at ``/var/run/gco/ca`` -- 

13the client container never sees a private key, and no extra trust 

14distribution machinery is needed. 

15 

16Contexts are cached per CA file identity (path, inode, modification time, 

17size). Kubernetes updates a projected Secret by swapping a symlink, which 

18changes the resolved inode and mtime, so a rotated CA is picked up by the 

19next context request without restarting the process; long-lived clients 

20that captured a context keep it until they are rebuilt. 

21""" 

22 

23from __future__ import annotations 

24 

25import os 

26import ssl 

27import threading 

28from pathlib import Path 

29from urllib.parse import urlsplit 

30 

31INTERNAL_CA_FILE_ENV = "GCO_INTERNAL_CA_FILE" 

32DEFAULT_INTERNAL_CA_FILE = "/var/run/gco/ca/ca.crt" 

33 

34_CacheKey = tuple[str, int, int, int] 

35_context_cache: dict[str, tuple[_CacheKey, ssl.SSLContext]] = {} 

36_cache_lock = threading.Lock() 

37 

38 

39class InternalTLSError(RuntimeError): 

40 """The GCO internal CA bundle is missing, unreadable, or not a usable CA.""" 

41 

42 

43def internal_ca_file() -> Path: 

44 """Return the CA bundle path: ``GCO_INTERNAL_CA_FILE`` or the projected default.""" 

45 configured = os.getenv(INTERNAL_CA_FILE_ENV, "").strip() 

46 return Path(configured or DEFAULT_INTERNAL_CA_FILE) 

47 

48 

49def _cache_key(path: Path) -> _CacheKey: 

50 try: 

51 stat = path.stat() 

52 except OSError as exc: 

53 raise InternalTLSError( 

54 f"GCO internal CA bundle is not readable at {path}: {exc.strerror or exc}" 

55 ) from exc 

56 return (str(path), stat.st_ino, stat.st_mtime_ns, stat.st_size) 

57 

58 

59def _build_context(path: Path) -> ssl.SSLContext: 

60 """Client context that trusts exactly the CA bundle at ``path``.""" 

61 try: 

62 # With an explicit cafile the default context loads no system anchors: 

63 # the internal CA is the only trust root. 

64 context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=str(path)) 

65 except (OSError, ssl.SSLError) as exc: 

66 raise InternalTLSError( 

67 f"GCO internal CA bundle at {path} is not a usable CA: {exc}" 

68 ) from exc 

69 context.minimum_version = ssl.TLSVersion.TLSv1_2 

70 return context 

71 

72 

73def internal_ssl_context(ca_file: str | os.PathLike[str] | None = None) -> ssl.SSLContext: 

74 """Return a verifying client context that trusts only the GCO internal CA. 

75 

76 Hostname verification stays on (the default for ``Purpose.SERVER_AUTH``), 

77 so the peer must present a certificate for the exact Service name dialled. 

78 Raises :class:`InternalTLSError` when the bundle is missing or invalid, so 

79 callers fail closed instead of silently falling back to another trust store. 

80 """ 

81 path = Path(ca_file) if ca_file is not None else internal_ca_file() 

82 key = _cache_key(path) 

83 with _cache_lock: 

84 cached = _context_cache.get(key[0]) 

85 if cached is not None and cached[0] == key: 

86 return cached[1] 

87 context = _build_context(path) 

88 with _cache_lock: 

89 _context_cache[key[0]] = (key, context) 

90 return context 

91 

92 

93def verify_for_url( 

94 url: str, ca_file: str | os.PathLike[str] | None = None 

95) -> ssl.SSLContext | bool: 

96 """Return the httpx2 ``verify=`` value for an in-cluster URL. 

97 

98 ``https`` URLs verify against the internal CA. Any other scheme -- a local 

99 or CI override such as ``http://127.0.0.1:8080`` -- opens no TLS session, 

100 so the library default is returned and no CA bundle is required. 

101 """ 

102 if urlsplit(url).scheme.lower() == "https": 

103 return internal_ssl_context(ca_file) 

104 return True 

105 

106 

107def clear_cache() -> None: 

108 """Forget cached contexts (tests, and callers that rebuild after an error).""" 

109 with _cache_lock: 

110 _context_cache.clear()