Kubernetesクラスタへのアクセス権限をブートストラップする

//+kubebuilder:rbac:groups=cr.requests.test,resources=accessrequests,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cr.requests.test,resources=accessrequests/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cr.requests.test,resources=accessrequests/finalizers,verbs=update
//+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=role,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=rolebinding,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=,resources=serviceaccount,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=,resources=secret,verbs=get;list;watch;create;update;patch;delete

func (r *AccessRequestReconciler) SetupWithManager(mgr ctrl.Manager) error {
        return ctrl.NewControllerManagedBy(mgr).
                For(&crv1beta1.AccessRequest{}).
                Owns(&corev1.ServiceAccount{}).
                Complete(r)
}

const (
        defaultNamespace   = "default"
        defaultRoleBinding = "pod-viewer"
        controllerKey      = "controller"
        controllerVal      = "AccessRequest"
)
 
var (
        defaultVerb = [...]string{"get", "list", "watch"}
        letters     = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
)

func (r *AccessRequestReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
        l := log.FromContext(ctx)
 
        var (
                err   error
                ca    string
                token string
                obj   crv1beta1.AccessRequest
                sa    *corev1.ServiceAccount
                sec   corev1.Secret
        )
 
        if err = r.Get(ctx, req.NamespacedName, &obj); err != nil {
                if apiErrors.IsNotFound(err) {
                        return ctrl.Result{}, nil
                }
                l.Info(err.Error())
                return ctrl.Result{}, err
        }
 
        if !obj.DeletionTimestamp.IsZero() {
                return ctrl.Result{}, nil
        }
 
        if a := obj.Status.Accepted; !a {
                fmt.Println("not accepteed")
                return ctrl.Result{}, nil
        }
 
        sa, _, err = r.reconcileServiceAccount(ctx, &obj)
        if len(sa.Secrets) == 0 {
                return ctrl.Result{RequeueAfter: time.Second * 2}, nil
        }
        sec = corev1.Secret{}
        sec.Name = sa.Secrets[0].Name
        sec.Namespace = sa.Namespace
 
        if err = r.Get(ctx, client.ObjectKeyFromObject(&sec), &sec); err != nil {
                return ctrl.Result{}, err
        }
 
        key := generateKey(32)
        ekey, err := encryptRSA(obj.Spec.PubKey, []byte(key))
        ca, err = encryptAES(key, sec.Data["ca.crt"])
        if err != nil {
                return ctrl.Result{}, err
        }
        token, err = encryptAES(key, sec.Data["token"])
        if err != nil {
                return ctrl.Result{}, err
        }
        obj.Status.CA = ca
        obj.Status.Key = ekey
        obj.Status.Token = token
        obj.Status.Ready = true
        err = r.Status().Update(ctx, &obj)
        if err != nil {
                return ctrl.Result{}, err
        }
 
        if err = r.reconcileRoleBinding(ctx, &obj); err != nil {
                return ctrl.Result{}, err
        }
 
        return ctrl.Result{}, nil
}

func (r *AccessRequestReconciler) reconcileServiceAccount(ctx context.Context, req *crv1beta1.AccessRequest) (*corev1.ServiceAccount, bool, error) {
        var (
                err error
        )
        sa := corev1.ServiceAccount{}
        sa.Name = req.Name
        sa.Namespace = defaultNamespace
        if err = r.Get(ctx, client.ObjectKeyFromObject(&sa), &sa); err != nil {
                if client.IgnoreNotFound(err) != nil {
                        return nil, false, err
                }
                sa.Annotations = map[string]string{controllerKey: controllerVal}
                if err = r.Create(ctx, &sa); err != nil {
                        return nil, false, err
                }
                return &sa, true, nil
        }
 
        cn, found := sa.Annotations[controllerKey]
        if !found {
                return nil, false, errors.New("ServiceAccount is not under control")
        }
        if cn != controllerVal {
                return nil, false, errors.New("ServiceAccount has other controller")
        }
        return &sa, false, nil
}

func (r *AccessRequestReconciler) reconcileRoleBinding(ctx context.Context, req *crv1beta1.AccessRequest) error {
        var err error
        rb := rbacv1.RoleBinding{}
        rb.Name = defaultRoleBinding
        rb.Namespace = defaultNamespace
        if err = r.Get(ctx, client.ObjectKeyFromObject(&rb), &rb); err != nil {
                return err
        }
 
        s := rbacv1.Subject{
                Kind:      "ServiceAccount",
                Namespace: defaultNamespace,
                Name:      req.Name,
        }
        if !contains(rb.Subjects, s) {
                rb.Subjects = append(rb.Subjects, s)
        }
 
        if err = r.Update(ctx, &rb); err != nil {
                return err
        }
        return nil
}
 
func contains(ss []rbacv1.Subject, s rbacv1.Subject) bool {
        for _, cs := range ss {
                if cs.Kind == s.Kind &&
                        cs.Name == s.Name &&
                        cs.Namespace == s.Namespace {
                        return true
                }
        }
        return false
}

func encryptRSA(k string, s []byte) (string, error) {
        pkey, _, _, _, err := ssh.ParseAuthorizedKey([]byte(k))
        if err != nil {
                return "", err
        }
        pckey := pkey.(ssh.CryptoPublicKey)
        prkey := pckey.CryptoPublicKey().(*rsa.PublicKey)
        enc, err := rsa.EncryptOAEP(sha512.New(), crand.Reader, prkey, s, nil)
        if err != nil {
                return "", err
        }
        emsg := base64.StdEncoding.EncodeToString(enc)
        return emsg, nil
}
 
func encryptAES(k string, s []byte) (string, error) {
        c, err := aes.NewCipher([]byte(k))
        if err != nil {
                return "", err
        }
 
        gcm, err := cipher.NewGCM(c)
        if err != nil {
                return "", err
        }
 
        nonce := make([]byte, gcm.NonceSize())
 
        if _, err = io.ReadFull(crand.Reader, nonce); err != nil {
                return "", err
        }
 
        enc := gcm.Seal(nonce, nonce, s, nil)
        emsg := base64.StdEncoding.EncodeToString(enc)
        return emsg, nil
}
 
func generateKey(n int) string {
        b := make([]rune, n)
        for i := range b {
                b[i] = letters[mrand.Intn(len(letters))]
        }
        return string(b)
}

import time
import socket
import subprocess
import string
import random
import binascii
import base64
 
from Crypto.PublicKey import RSA
from Crypto.Cipher import PKCS1_OAEP, AES
from Crypto.Hash import SHA512, SHA1
from Crypto.Signature import pss
 
import requests
 
import click
 
from kubernetes import client, config
 
run = subprocess.getoutput
 
 
def generate_sshkey(keyname):
    cmd = f'ssh-keygen -f {keyname} -N "" <<<y >/dev/null 2>&1'
    run(cmd)
 
    with open(f"{keyname}", 'r') as f:
        data = f.readlines()
    rkey = RSA.import_key(''.join(data))
 
    with open(f"{keyname}.pub", 'r') as f:
        data = f.readlines()
    pkey = RSA.import_key(''.join(data))
    return rkey, pkey
 
 
def create_access_request(url, path, pkey):
    cr = {
            'apiVersion': 'cr.requests.test/v1beta1',
            'kind': 'AccessRequest',
            'metadata': {},
            'spec': {},
            }
 
    cr['metadata']['name'] = socket.gethostname() + "-" + ''.join(random.choice(string.ascii_lowercase + string.digits) for _ in range(5))
    cr['spec']['pubKey'] = pkey.exportKey(format="OpenSSH").decode()
 
    try:
        req = requests.post(url+path, json=cr, verify=False)
    except:
        pass
 
    while True:
        req = requests.get(url+path+f"/{cr['metadata']['name']}", verify=False)
        ready = req.json().get('status', {}).get('ready', False)
        if ready:
            data = req.json()
            ca = data['status']['ca']
            token = data['status']['token']
            key = data['status']['key']
            print('access request accepted')
            break
        print('condition not met, waiting...')
        time.sleep(5)
    return ca, token, key
 
 
def decrypt_oaep(private_key, aes_ekey):
    d = PKCS1_OAEP.new(private_key, hashAlgo=SHA512, mgfunc=lambda x,y: pss.MGF1(x,y,SHA512))
    data = base64.b64decode(aes_ekey)
    m = d.decrypt(data)
    return m
 
 
def decrypt_aes(aes_key, edata):
    data = base64.b64decode(edata)
    nonce, emsg, tag = data[:12], data[12:-16], data[-16:]
 
    d = AES.new(aes_key, AES.MODE_GCM, nonce=nonce)
    m = d.decrypt_and_verify(emsg, tag)
    return m
 
 
def generate_kubeconfig(url, ca, token):
    template = '''apiVersion: v1
kind: Config
contexts:
- context:
    cluster: cluster
    user: pod-viewer
  name: pod-viewer
current-context: pod-viewer
clusters:
- cluster:
    certificate-authority-data: {ca}
    server: {server}
  name: cluster
users:
- name: pod-viewer
  user:
    token: {token}
'''
 
    kubeconfig = template.format(
            token=token.decode(),
            ca=base64.b64encode(ca).decode(),
            server=url
            )
    fkubeconfig = "gen.kubeconfig"
    with open(fkubeconfig, 'w') as f:
        f.write(kubeconfig)
    return fkubeconfig
 
 
def watch_pods(fkubeconfig):
    config.load_kube_config(fkubeconfig)
    v1 = client.CoreV1Api()
    while True:
        ret = v1.list_namespaced_pod("default")
        print('----')
        for po in ret.items:
            print(f"{po.metadata.namespace} {po.metadata.name} {po.status.pod_ip}")
        time.sleep(5)
 
 
@click.command()
@click.option("--url", type=str, help="kubernetes cluster api server url", default="https://localhost:6443")
@click.option("--keyname", type=str, help="ssh key name", default="test")
def main(url, keyname):
    path = "/apis/cr.requests.test/v1beta1/accessrequests"
 
    rkey, pkey = generate_sshkey(keyname)
    eca, etoken, ekey = create_access_request(url, path, pkey)
 
    aes_key = decrypt_oaep(rkey, ekey)
 
    token = decrypt_aes(aes_key, etoken)
    ca = decrypt_aes(aes_key, eca)
 
    fkubeconfig = generate_kubeconfig(url, ca, token)
    watch_pods(fkubeconfig)
 
 
if __name__ == "__main__":
    main()

動作確認の一例

試しに手元でkindでKubernetesクラスタを建てて、コントローラを動かしてみます。クライアントを動かすとAccessRequestリソースを作成して、Ready待ち状態になります。

人が承認としてAccessRequestリソースの.status.accepted: trueを記入すると、コントローラによって認証情報がAccessRequestリソースに置かれて、クライアント側で検知します。

あとはひたすらPodの情報を表示し続けます。サンプル例では途中でレプリカ数を増やして、その際のPodの追加やPodIPの割当が進んでいる状況が見れています。

$ python main.py --url https://127.0.0.1:42529
condition not met, waiting...   <------AccessRequestを作成して承認待ち状態に
condition not met, waiting...
condition not met, waiting...
condition not met, waiting...
condition not met, waiting...
access request accepted         <------AccessRequestがReadyになったのでkubeconfigを生成する
----
default curl-7bb98d65f4-gsj2f 10.244.1.8      <-------Kubernetesの情報が取得できるようになる
----
default curl-7bb98d65f4-gsj2f 10.244.1.8
----
default curl-7bb98d65f4-gsj2f 10.244.1.8
----
default curl-7bb98d65f4-6qcbz 10.244.2.8      <-------Deploymentのreplica数を変えて見ているのが観測できる
default curl-7bb98d65f4-ggnjc 10.244.1.9
default curl-7bb98d65f4-gsj2f 10.244.1.8
default curl-7bb98d65f4-xf7sf None
----
default curl-7bb98d65f4-6qcbz 10.244.2.8
default curl-7bb98d65f4-ggnjc 10.244.1.9
default curl-7bb98d65f4-gsj2f 10.244.1.8
default curl-7bb98d65f4-xf7sf 10.244.2.9
^C
Aborted!

まとめ

上記のようにコントローラを作成すると、kubeletが動いていないような場所でもKubernetes内の特定のリソースを見るように実装することができます。

応用としてVMをデプロイするとcloud-initでTokenを要請するようにして、Kubernetes側で承認するとkubeadm joinするためのTokenが受け渡されてNodeを追加するような仕組みも作れます。

また発展編としてAdmission Webhookを利用して、例えば同一L2セグメント内のAccessRequestsだけを許すようにすればよりセキュリティを高めることも可能です。

// 所感、言語が異なるものを使うと暗号化ライブラリ周りが結構ハマったのが辛かったです。