121 lines
4.8 KiB
Python
121 lines
4.8 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
from typing import List
|
|
|
|
import jinja2
|
|
from zope import component
|
|
from zope.interface import implementer
|
|
|
|
from ceo_common.interfaces import IConfig, IKubernetesService
|
|
from ceo_common.logger_factory import logger_factory
|
|
|
|
logger = logger_factory(__name__)
|
|
|
|
|
|
@implementer(IKubernetesService)
|
|
class KubernetesService:
|
|
namespace_prefix = 'csc-'
|
|
|
|
def __init__(self):
|
|
cfg = component.getUtility(IConfig)
|
|
self.members_clusterrole = cfg.get('k8s_members_clusterrole')
|
|
self.members_group = cfg.get('k8s_members_group')
|
|
self.authority_cert_path = cfg.get('k8s_authority_cert_path')
|
|
self.server_url = cfg.get('k8s_server_url')
|
|
self.jinja_env = jinja2.Environment(
|
|
loader=jinja2.PackageLoader('ceod.model'),
|
|
keep_trailing_newline=True,
|
|
)
|
|
|
|
def _run(self, args: List[str], check=True, **kwargs) -> subprocess.CompletedProcess:
|
|
return subprocess.run(args, check=check, text=True, **kwargs)
|
|
|
|
def _apply_manifest(self, manifest: str):
|
|
self._run(['kubectl', 'apply', '-f', '-'], input=manifest)
|
|
|
|
@classmethod
|
|
def _get_namespace(cls, username: str) -> str:
|
|
return cls.namespace_prefix + username
|
|
|
|
@classmethod
|
|
def _get_username_from_namespace(cls, namespace: str) -> str:
|
|
assert namespace.startswith(cls.namespace_prefix)
|
|
return namespace[len(cls.namespace_prefix):]
|
|
|
|
def create_account(self, username: str) -> str:
|
|
with tempfile.TemporaryDirectory() as tempdir:
|
|
# Create a new CSR
|
|
csr_path = os.path.join(tempdir, username + '.csr')
|
|
key_path = os.path.join(tempdir, username + '.key')
|
|
self._run([
|
|
'openssl', 'req', '-new', '-newkey', 'rsa:2048', '-nodes',
|
|
'-keyout', key_path, '-subj', f'/CN={username}/O={self.members_group}',
|
|
'-out', csr_path,
|
|
], stdin=subprocess.DEVNULL)
|
|
# Upload the CSR
|
|
encoded_csr = base64.b64encode(open(csr_path, 'rb').read()).decode()
|
|
csr_name = 'csc-' + username + '-csr'
|
|
template = self.jinja_env.get_template('kubernetes_csr.yaml.j2')
|
|
body = template.render(csr_name=csr_name, encoded_csr=encoded_csr)
|
|
self._apply_manifest(body)
|
|
# Approve the CSR
|
|
self._run(['kubectl', 'certificate', 'approve', csr_name])
|
|
# Wait until the certificate is issued
|
|
encoded_cert = ''
|
|
max_tries = 5
|
|
for i in range(max_tries):
|
|
proc = self._run([
|
|
'kubectl', 'get', 'csr', csr_name,
|
|
'-o', 'jsonpath={.status.certificate}',
|
|
], capture_output=True)
|
|
encoded_cert = proc.stdout
|
|
if encoded_cert != '':
|
|
break
|
|
time.sleep(1)
|
|
if encoded_cert == '':
|
|
raise Exception('Timed out waiting for certificate to get issued')
|
|
# Delete the CSR
|
|
self._run(['kubectl', 'delete', 'csr', csr_name])
|
|
|
|
# Create a namespace
|
|
namespace = self._get_namespace(username)
|
|
template = self.jinja_env.get_template('kubernetes_user.yaml.j2')
|
|
body = template.render(
|
|
username=username, namespace=namespace,
|
|
members_clusterrole=self.members_clusterrole)
|
|
self._apply_manifest(body)
|
|
|
|
# Return the kubeconfig
|
|
encoded_key = base64.b64encode(open(key_path, 'rb').read()).decode()
|
|
encoded_authority_cert = base64.b64encode(
|
|
open(self.authority_cert_path, 'rb').read()
|
|
).decode()
|
|
template = self.jinja_env.get_template('kubeconfig.j2')
|
|
body = template.render(
|
|
username=username, namespace=namespace,
|
|
server_url=self.server_url,
|
|
encoded_cert=encoded_cert, encoded_key=encoded_key,
|
|
encoded_authority_cert=encoded_authority_cert)
|
|
return body
|
|
|
|
def delete_account(self, username: str):
|
|
logger.info(f'Deleting Kubernetes namespace for {username}')
|
|
namespace = self._get_namespace(username)
|
|
# don't check exit code because namespace might not exist
|
|
self._run(['kubectl', 'delete', 'namespace', namespace], check=False)
|
|
|
|
def get_accounts(self) -> List[str]:
|
|
proc = self._run(['kubectl', 'get', 'namespaces', '-o', 'json'],
|
|
capture_output=True)
|
|
items = json.loads(proc.stdout)['items']
|
|
namespaces = [item['metadata']['name'] for item in items]
|
|
return [
|
|
self._get_username_from_namespace(namespace)
|
|
for namespace in namespaces
|
|
if namespace.startswith(self.namespace_prefix)
|
|
]
|