informed-decision/informed_decision/workload.py

"""Private workload composition. Kubernetes verifies identity; PDP grants actions."""
import json
import math
from pathlib import Path
import re
import ssl
import time

import jwt

from .disposition import Actor, ActorKind
from .http_transport import JSONTransport, https_origin
from .intake_service import IntakeError
from .lifecycle import Attribution
from .lifecycle_http import LifecycleAPI
from .lifecycle_service import Caller
from .records import dumps

LIFECYCLE_AUDIENCE = 'informed-decision-lifecycle'
SA = re.compile(r'system:serviceaccount:[a-z0-9-]+:[a-z0-9-]+')


def registry(path):
    with Path(path).open('rb') as handle:
        raw = handle.read(65537)
    if len(raw) > 65536:
        raise ValueError('registry too large')
    result = json.loads(raw)
    if (not isinstance(result, dict) or any(not SA.fullmatch(name) or
            not isinstance(row, dict) or set(row) != {'tenant'} or row['tenant'] != 'tenant:platform'
            for name, row in result.items())):
        raise ValueError('invalid workload registry')
    return result


class KubernetesTokenReview:
    def __init__(self, origin, ca_file, token_provider, *, transport_factory=JSONTransport):
        self.origin = https_origin(origin)
        self.ca_file = Path(ca_file)
        if not self.ca_file.is_absolute():
            raise ValueError('absolute Kubernetes CA path required')
        self.token_provider = token_provider
        self.transport_factory = transport_factory

    def __call__(self, body):
        try:
            # Reopen projected CA/caller custody; require TLS peer and hostname validation.
            context = ssl.create_default_context(cafile=str(self.ca_file))
            transport = self.transport_factory(ssl_context=context)
            status, result = transport.request('POST', self.origin + '/apis/authentication.k8s.io/v1/tokenreviews',
                headers={'Authorization': 'Bearer ' + self.token_provider(), 'Content-Type': 'application/json'},
                body=dumps(body).encode())
            if status not in (200, 201):
                raise ValueError()
            return result
        except Exception:
            raise IntakeError(503, 'workload_identity_unavailable') from None


class LifecycleWorkloadIdentity:
    def __init__(self, review, callers_file, *, clock=time.time):
        self.review, self.callers_file, self.clock = review, Path(callers_file), clock
        if not self.callers_file.is_absolute():
            raise ValueError('absolute workload registry path required')

    def __call__(self, token):
        if not isinstance(token, str) or not 1 <= len(token) <= 16384:
            raise ValueError('invalid workload token')
        result = self.review({'apiVersion': 'authentication.k8s.io/v1', 'kind': 'TokenReview',
                              'spec': {'token': token, 'audiences': [LIFECYCLE_AUDIENCE]}})
        status = result['status']
        user = status['user']
        if (status.get('authenticated') is not True or status.get('error')
                or LIFECYCLE_AUDIENCE not in status.get('audiences', [])
                or not isinstance(user.get('username'), str) or not SA.fullmatch(user['username'])
                or not isinstance(user.get('uid'), str) or not user['uid']):
            raise ValueError('workload authentication refused')
        admitted = registry(self.callers_file).get(user['username'])
        if admitted is None:
            raise ValueError('workload not admitted')
        # The API server has verified this Kubernetes JWT. Read only its expiry;
        # subject/UID come from TokenReview and tenant from reviewed registration.
        claims = jwt.decode(token, options={'verify_signature': False})
        expiry = claims.get('exp')
        if type(expiry) is not int or not math.isfinite(expiry) or expiry <= self.clock():
            raise ValueError('workload token expired')
        return Caller(Attribution(admitted['tenant'], Actor(user['username'], ActorKind.SERVICE)),
                      min(expiry, self.clock() + 30), dumps({
                          'principal_type_source': 'authentication-derived',
                          'tenant_source': 'registration-supplied',
                          'workload_uid': user['uid'], 'authentication_audience': LIFECYCLE_AUDIENCE}))


class PrivateWorkloadApp:
    def __init__(self, *, intake=None, lifecycle=None):
        self.intake, self.lifecycle = intake, lifecycle

    def __call__(self, environ, start_response):
        path = environ.get('PATH_INFO', '')
        if path == '/v1/memos' and self.intake is not None:
            return self.intake(environ, start_response)
        if (path == '/v1/decisions' or path.startswith('/v1/decisions/')) and self.lifecycle is not None:
            return self.lifecycle(environ, start_response)
        raw = b'{"error":"route_not_found"}'
        start_response('404 Not Found', [('Content-Type', 'application/json'), ('Cache-Control', 'no-store'),
                                        ('Content-Length', str(len(raw)))])
        return [raw]


def compose_workload(data, store, lifecycle, approval_origin, token_reader):
    if (not isinstance(data, dict) or set(data) - {'token_review', 'intake', 'lifecycle_callers_file'}
            or 'token_review' not in data or not ({'intake', 'lifecycle_callers_file'} & data.keys())):
        raise ValueError('invalid private workload configuration')
    review = data['token_review']
    if not isinstance(review, dict) or set(review) != {'origin', 'ca_file', 'token_file'}:
        raise ValueError('exact TokenReview configuration required')
    verifier = KubernetesTokenReview(review['origin'], review['ca_file'], token_reader(review['token_file']))
    intake, api = None, None
    if 'intake' in data:
        from .requester import compose_intake
        configuration = data['intake']
        if not isinstance(configuration, dict) or set(configuration) != {'secret_file', 'submitters_file', 'groups_file'}:
            raise ValueError('invalid private intake configuration')
        if any(not isinstance(p, str) or not Path(p).is_absolute() for p in configuration.values()):
            raise ValueError('absolute intake projection paths required')
        intake = compose_intake(store, verifier, issuer='https://kc.coulomb.social',
            secret_file=configuration['secret_file'], approval_origin=approval_origin,
            submitters_path=configuration['submitters_file'], groups_path=configuration['groups_file'])
    if 'lifecycle_callers_file' in data:
        if lifecycle is None:
            raise ValueError('private lifecycle API requires a separately pinned policy')
        api = LifecycleAPI(lifecycle, LifecycleWorkloadIdentity(verifier, data['lifecycle_callers_file']))
    return PrivateWorkloadApp(intake=intake, lifecycle=api)