Example of a policy loading operator for K8s and Conjur

Readme.md

Requirements

In order to run this operator you need to install kopf, requests, and conjur-api which technically requires python3.10. However the PyPI install is broken anyway, so if you download and install the module from source you can change the required version in the setup.cfg file to the python3 version you have. As long as it was compiled with openssl 1.1.1 it should work.

You can install the first two requirements with python3 -m pip install kopf requests and the downloaded/cloned version of conjur-api with python3 -m pip install -e . while in the project directory.

After this you will need a running instance of Conjur. The file is currently setup to connect to the developer version in the cyberark/conjur repo. You will need update the password variable with the correct API key no matter what, and the other conjur configuration variables if you use a different Conjur instance.

After this you can run the operator with kopf example_operator.py and create a new namespace in your cluster with the prefix conj- and a new policy should be added to the conjur instance in root.

example_operator.py

#!/usr/bin/env python3
import tempfile
import os

import kopf
import requests
from conjur_api.models import *
from conjur_api import *
from conjur_api.providers.simple_credentials_provider import SimpleCredentialsProvider

def get_conjur_client():
    conjur_url = 'http://localhost:3000'
    account = 'cucumber'
    username = 'admin'
    password = '3p3585mf8t0b32qxfgy422eecsq15afyz71pqgzr52b21j051be2tm6'

    connection_info = ConjurConnectionInfo(conjur_url, account, None)
    creds = CredentialsData(username=username, password=password, machine=conjur_url)

    provider = SimpleCredentialsProvider()
    provider.save(creds)

    return Client(connection_info, credentials_provider=provider, ssl_verification_mode=SslVerificationMode.INSECURE, async_mode=False)

def get_policy_file(host_name):
    # For some reason the conjur python API doesn't provide faculties for passing a string
    # as policy text so we need to create a tempfile and populate with our policy yaml

    file = tempfile.NamedTemporaryFile(mode='w+', delete=False)
    # Could easily add more to this policy file but just keep it simple for this example
    file.write(f'- !host {host_name}\n')
    file.close()
    return file.name

@kopf.on.create('namespaces')
def create_fn(body, **kwargs):
    # Gets the name of the new namespace we just added from the body of the k8s api request
    name = body['metadata']['name']

    # If this isn't a conjur attached namespace quit
    if name[:5] != "conj-":
        return

    client = get_conjur_client()
    policy_file = get_policy_file(name)
    client.login()
    # Just load the policy into root for now
    print(client.load_policy_file('root', policy_file))
    os.unlink(policy_file)