Merge pull request #721 from nolar/webhook-auto-servers-tunnels

Auto-detect cluster type for proper webhook server's hostname

docs/admission.rst

@@ -354,10 +354,17 @@ each with its configuration parameters (see their descriptions):
 usually to a locally running *webhook server*.
 
 * `kopf.WebhookNgrokTunnel` established a tunnel through ngrok_.
-* `kopf.WebhookInletsTunnel` tunnels the traffic through inlets_.
 
 .. _ngrok: https://ngrok.com/
-.. _inlets: https://inlets.dev/
+
+For ease of use, the cluster type can be recognised automatically in some cases:
+
+* `kopf.WebhookAutoServer` runs locally, detects Minikube & K3s, and uses them
+  via their special hostnames. If it cannot detect the cluster type, it runs
+  a simple local webhook server. The auto-server never tunnels.
+* `kopf.WebhookAutoTunnel` attempts to use an auto-server if possible.
+  If not, it uses one of the available tunnels (currently, only ngrok).
+  This is the most universal way to make any environment work.
 
 .. note::
     External tunnelling services usually limit the number of requests.
@@ -369,7 +376,7 @@ usually to a locally running *webhook server*.
 
 .. note::
     A reminder: using development-mode tunnels and self-signed certificates
-    requires an extra: ``pip install kopf[dev]``.
+    requires extra dependencies: ``pip install kopf[dev]``.
 
 
 Authenticate apiservers
@@ -545,7 +552,7 @@ do not support HTTPS tunnelling (or require paid subscriptions):
 
     @kopf.on.startup()
     def config(settings: kopf.OperatorSettings, **_):
-        settings.admission.server = kopf.WebhookServer(insecure=True)
+        settings.admission.server = kopf.¶ver(insecure=True)
 
 
 Custom servers/tunnels

examples/17-admission/example.py

@@ -1,41 +1,61 @@
 import pathlib
-from typing import Dict
+from typing import Dict, List
 
 import kopf
 
+ROOT = (pathlib.Path.cwd() / pathlib.Path(__file__)).parent.parent.parent
+
 
 @kopf.on.startup()
 def config(settings: kopf.OperatorSettings, **_):
-    ROOT = (pathlib.Path.cwd() / pathlib.Path(__file__)).parent.parent.parent
-    settings.admission.managed = 'auto.kopf.dev'
+
+    # Plain and simple local endpoint with an auto-generated certificate:
+    settings.admission.server = kopf.WebhookServer()
+
+    # Plain and simple local endpoint with with provided certificate (e.g. openssl):
+    settings.admission.server = kopf.WebhookServer(certfile=ROOT/'cert.pem', pkeyfile=ROOT/'key.pem', port=1234)
+
+    # K3d/K3s-specific server that supports accessing from inside of a VM (a generated certificate):
     settings.admission.server = kopf.WebhookK3dServer(cadump=ROOT/'ca.pem')
-    ## Other options (see the docs):
-    # settings.admission.server = kopf.WebhookServer()
-    # settings.admission.server = kopf.WebhookServer(certfile=ROOT/'cert.pem', pkeyfile=ROOT/'key.pem', port=1234)
-    # settings.admission.server = kopf.WebhookK3dServer(cadump=ROOT/'ca.pem')
-    # settings.admission.server = kopf.WebhookK3dServer(certfile=ROOT/'k3d-cert.pem', pkeyfile=ROOT/'k3d-key.pem', port=1234)
-    # settings.admission.server = kopf.WebhookMinikubeServer(port=1234, cadump=ROOT/'ca.pem', verify_cafile=ROOT/'client-cert.pem')
-    # settings.admission.server = kopf.WebhookNgrokTunnel()
-    # settings.admission.server = kopf.WebhookNgrokTunnel(binary="/usr/local/bin/ngrok", token='...', port=1234)
-    # settings.admission.server = kopf.WebhookNgrokTunnel(binary="/usr/local/bin/ngrok", port=1234, path='/xyz', region='eu')
+
+    # K3d/K3s-specific server that supports accessing from inside of a VM (a provided certificate):
+    settings.admission.server = kopf.WebhookK3dServer(certfile=ROOT/'k3d-cert.pem', pkeyfile=ROOT/'k3d-key.pem', port=1234)
+
+    # Minikube-specific server that supports accessing from inside of a VM (a generated certificate):
+    settings.admission.server = kopf.WebhookMinikubeServer(port=1234, cadump=ROOT/'ca.pem')
+
+    # Tunneling Kubernetes->ngrok->local server (anonymous, auto-loaded binary):
+    settings.admission.server = kopf.WebhookNgrokTunnel(path='/xyz', port=1234)
+
+    # Tunneling Kubernetes->ngrok->local server (registered users, pre-existing binary):
+    settings.admission.server = kopf.WebhookNgrokTunnel(binary="/usr/local/bin/ngrok", token='...', )
+
+    # Tunneling Kubernetes->ngrok->local server (registered users, pre-existing binary, specific region):
+    settings.admission.server = kopf.WebhookNgrokTunnel(binary="/usr/local/bin/ngrok", region='eu')
+
+    # Auto-detect the best server (K3d/Minikube/simple) strictly locally:
+    settings.admission.server = kopf.WebhookAutoServer()
+
+    # Auto-detect the best server (K3d/Minikube/simple) with external tunneling as a fallback:
+    settings.admission.server = kopf.WebhookAutoTunnel()
+
+    # The final configuration for CI/CD (overrides previous values):
+    settings.admission.server = kopf.WebhookAutoServer()
+    settings.admission.managed = 'auto.kopf.dev'
 
 
 @kopf.on.validate('kex')
-def authhook(headers, sslpeer, warnings, **_):
-    # print(f'headers={headers}')
-    # print(f'sslpeer={sslpeer}')
-    if not sslpeer:
+def authhook(headers: kopf.Headers, sslpeer: kopf.SSLPeer, warnings: List[str], **_):
+    user_agent = headers.get('User-Agent', '(unidentified)')
+    warnings.append(f"Accessing as user-agent: {user_agent}")
+    if not sslpeer.get('subject'):
         warnings.append("SSL peer is not identified.")
     else:
-        common_name = None
-        for key, val in sslpeer['subject'][0]:
-            if key == 'commonName':
-                common_name = val
-                break
+        common_names = [val for key, val in sslpeer['subject'][0] if key == 'commonName']
+        if common_names:
+            warnings.append(f"SSL peer is {common_names[0]}.")
         else:
             warnings.append("SSL peer's common name is absent.")
-        if common_name is not None:
-            warnings.append(f"SSL peer is {common_name}.")
 
 
 @kopf.on.validate('kex')

kopf/__init__.py

@@ -160,6 +160,8 @@
     WebhookK3dServer,
     WebhookMinikubeServer,
     WebhookNgrokTunnel,
+    WebhookAutoServer,
+    WebhookAutoTunnel,
 )
 from kopf.utilities.piggybacking import (
     login_via_pykube,
@@ -194,6 +196,8 @@
     'WebhookK3dServer',
     'WebhookMinikubeServer',
     'WebhookNgrokTunnel',
+    'WebhookAutoServer',
+    'WebhookAutoTunnel',
     'PermanentError',
     'TemporaryError',
     'HandlerTimeoutError',

kopf/clients/scanning.py

@@ -1,10 +1,42 @@
 import asyncio
-from typing import Collection, Optional, Set
+import ssl
+import urllib.parse
+from typing import Collection, Mapping, Optional, Set, Tuple
 
 from kopf.clients import auth, errors
 from kopf.structs import references
 
 
+@auth.reauthenticated_request
+async def read_sslcert(
+        *,
+        context: Optional[auth.APIContext] = None,
+) -> Tuple[str, bytes]:
+    if context is None:
+        raise RuntimeError("API instance is not injected by the decorator.")
+
+    parsed = urllib.parse.urlparse(context.server)
+    host = parsed.hostname or ''  # NB: it cannot be None/empty in our case.
+    port = parsed.port or 443
+    loop = asyncio.get_running_loop()
+    cert = await loop.run_in_executor(None, ssl.get_server_certificate, (host, port))
+    return host, cert.encode('ascii')
+
+
+@auth.reauthenticated_request
+async def read_version(
+        *,
+        context: Optional[auth.APIContext] = None,  # injected by the decorator
+) -> Mapping[str, str]:
+    if context is None:
+        raise RuntimeError("API instance is not injected by the decorator.")
+
+    server = context.server.rstrip('/')
+    url = f'{server}/version'
+    rsp: Mapping[str, str] = await errors.parse_response(await context.session.get(url))
+    return rsp
+
+
 @auth.reauthenticated_request
 async def scan_resources(
         *,

kopf/toolkits/webhooks.py

@@ -18,6 +18,7 @@
 
 import aiohttp.web
 
+from kopf.clients import scanning
 from kopf.reactor import admission
 from kopf.structs import reviews
 
@@ -535,3 +536,116 @@ async def __call__(self, fn: reviews.WebhookFn) -> AsyncIterator[reviews.Webhook
         finally:
             if tunnel is not None:
                 await loop.run_in_executor(None, ngrok.disconnect, tunnel.public_url)
+
+
+class ClusterDetector:
+    """
+    A mixing for auto-server/auto-tunnel to detect the cluster type.
+
+    The implementation of the server detection requires the least possible
+    permissions or no permissions at all. In most cases, it will identify
+    the server type by its SSL certificate meta-information (subject/issuer).
+    SSL information is the most universal way for all typical local clusters.
+
+    If SSL parsing fails, it will try to fetch the information from the cluster.
+    However, it rarely contains any useful information about the cluster's
+    surroundings and environment, but only about the cluster itself
+    (though it helps with K3s).
+
+    Note: the SSL certificate of the Kubernetes API is checked, not of webhooks.
+    """
+    @staticmethod
+    async def guess_host() -> Optional[str]:
+        try:
+            import certvalidator
+        except ImportError:
+            raise MissingDependencyError(
+                "Auto-guessing cluster types requires an extra dependency: "
+                "run `pip install certvalidator` or `pip install kopf[dev]`. "
+                "More: https://kopf.readthedocs.io/en/stable/admission/")
+
+        hostname, cert = await scanning.read_sslcert()
+        valcontext = certvalidator.ValidationContext(extra_trust_roots=[cert])
+        validator = certvalidator.CertificateValidator(cert, validation_context=valcontext)
+        certpath = validator.validate_tls(hostname)
+        issuer_cn = certpath.first.issuer.native.get('common_name', '')
+        subject_cn = certpath.first.subject.native.get('common_name', '')
+        subject_org = certpath.first.subject.native.get('organization_name', '')
+
+        if subject_cn == 'k3s' or subject_org == 'k3s' or issuer_cn.startswith('k3s-'):
+            return WebhookK3dServer.DEFAULT_HOST
+        elif subject_cn == 'minikube' or issuer_cn == 'minikubeCA':
+            return WebhookMinikubeServer.DEFAULT_HOST
+        else:
+            versioninfo = await scanning.read_version()
+            if '+k3s' in versioninfo.get('gitVersion', ''):
+                return WebhookK3dServer.DEFAULT_HOST
+        return None
+
+
+class WebhookAutoServer(ClusterDetector, WebhookServer):
+    """
+    A locally listening webserver which attempts to guess its proper hostname.
+
+    The choice is happening between supported webhook servers only
+    (K3d/K3d and Minikube at the moment). In all other cases,
+    a regular local server is started without hostname overrides.
+
+    If automatic tunneling is possible, consider `WebhookAutoTunnel` instead.
+    """
+    async def __call__(self, fn: reviews.WebhookFn) -> AsyncIterator[reviews.WebhookClientConfig]:
+        host = self.DEFAULT_HOST = await self.guess_host()
+        if host is None:
+            logger.debug(f"Cluster detection failed, running a simple local server.")
+        else:
+            logger.debug(f"Cluster detection found the hostname: {host}")
+        async for client_config in super().__call__(fn):
+            yield client_config
+
+
+class WebhookAutoTunnel(ClusterDetector):
+    """
+    The same as `WebhookAutoServer`, but with possible tunneling.
+
+    Generally, tunneling gives more possibilities to run in any environment,
+    but it must not happen without a permission from the developers,
+    and is not possible if running in a completely isolated/local/CI/CD cluster.
+    Therefore, developers should activated automatic setup explicitly.
+
+    If automatic tunneling is prohibited or impossible, use `WebhookAutoServer`.
+
+    .. note::
+
+        Automatic server/tunnel detection is highly limited in configuration
+        and provides only the most common options of all servers & tunners:
+        specifically, listening ``addr:port/path``.
+        All other options are specific to their servers/tunnels
+        and the auto-guessing logic cannot use/accept/pass them.
+    """
+    addr: Optional[str]  # None means "any interface"
+    port: Optional[int]  # None means a random port
+    path: Optional[str]
+
+    def __init__(
+            self,
+            *,
+            addr: Optional[str] = None,
+            port: Optional[int] = None,
+            path: Optional[str] = None,
+    ) -> None:
+        super().__init__()
+        self.addr = addr
+        self.port = port
+        self.path = path
+
+    async def __call__(self, fn: reviews.WebhookFn) -> AsyncIterator[reviews.WebhookClientConfig]:
+        server: reviews.WebhookServerProtocol
+        host = await self.guess_host()
+        if host is None:
+            logger.debug(f"Cluster detection failed, using an ngrok tunnel.")
+            server = WebhookNgrokTunnel(addr=self.addr, port=self.port, path=self.path)
+        else:
+            logger.debug(f"Cluster detection found the hostname: {host}")
+            server = WebhookServer(addr=self.addr, port=self.port, path=self.path, host=host)
+        async for client_config in server(fn):
+            yield client_config

setup.py

@@ -58,6 +58,7 @@
             'pyngrok',          # 1.00 MB + downloaded binary
             'oscrypto',         # 2.80 MB (smaller than cryptography: 8.7 MB)
             'certbuilder',      # +0.1 MB (2.90 MB if alone)
+            'certvalidator',    # +0.1 MB (2.90 MB if alone)
         ],
     },
     package_data={"kopf": ["py.typed"]},

tests/admission/conftest.py

@@ -3,6 +3,8 @@
 import warnings
 from unittest.mock import Mock
 
+import pyngrok.conf
+import pyngrok.ngrok
 import pytest
 
 from kopf.reactor.indexing import OperatorIndexers
@@ -139,3 +141,13 @@ def k8s_mocked(mocker):
         post_event=mocker.patch('kopf.clients.events.post_event'),
         sleep_or_wait=mocker.patch('kopf.structs.primitives.sleep_or_wait', return_value=None),
     )
+
+
+@pytest.fixture(autouse=True)
+def pyngrok_mock(mocker):
+    mocker.patch.object(pyngrok.conf, 'get_default')
+    mocker.patch.object(pyngrok.ngrok, 'set_auth_token')
+    mocker.patch.object(pyngrok.ngrok, 'connect')
+    mocker.patch.object(pyngrok.ngrok, 'disconnect')
+    pyngrok.ngrok.connect.return_value.public_url = 'https://nowhere'
+    return pyngrok