configmap.yaml

apiVersion: v1
data:
  controller.py: |
    import base64
    import copy
    import http
    import http.server as http_server
    import http.client as http_client
    import json
    import logging
    import os
    import pprint
    import re
    import string
    import secrets
    from urllib import parse

    ADMIN_USERNAME = os.environ["GITEA_ADMIN_USERNAME"]
    ADMIN_PASSWORD = os.environ["GITEA_ADMIN_PASSWORD"]
    BASE_URL = os.environ["GITEA_BASE_URL"]

    PARSED_BASE_URL = None
    try:
        PARSED_BASE_URL = parse.urlparse(BASE_URL)
    except Exception as e:
        raise RuntimeError(f"bad GITEA_BASE_URL '{BASE_URL}': {e}")

    tmp = PARSED_BASE_URL.path
    if len(tmp) > 1 and tmp[-1] == "/":
        tmp = tmp[:-1]
    API_BASE_URL = f"{tmp}/api/v1"


    def b64string(s):
        return base64.b64encode(s.encode("utf-8")).decode("utf-8")


    def from_b64string(s):
        return base64.b64decode(s.encode("utf-8")).decode("utf-8")


    credentials = b64string(f"{ADMIN_USERNAME}:{ADMIN_PASSWORD}")

    HEADERS = {
        "accept": "application/json",
        "content-type": "application/json",
        "authorization": f"Basic {credentials}",
    }

    PASSWORD_APLHABET = (
        string.ascii_letters + string.digits + string.punctuation + string.ascii_letters
    )
    PASSWORD_LENGTH = 12

    CASE_RX = re.compile(r"(?<!^)(?=[A-Z])")


    def create_result(state, message="awesome", children=[]):
        return {
            "status": {
                "state": state,
                "message": message,
            },
            "children": children,
        }


    def recase(x):
        if isinstance(x, list):
            return [recase(tmp) for tmp in x]

        if isinstance(x, tuple):
            return tuple([recase(tmp) for tmp in x])

        if not isinstance(x, dict):
            return x

        result = {}
        for k, v in x.items():
            k = CASE_RX.sub("_", k).lower()
            result[k] = recase(v)

        return result


    def do_request(method, endpoint, body=None):
        logging.info(f"connecting to: {PARSED_BASE_URL.scheme}://{PARSED_BASE_URL.netloc}")
        if PARSED_BASE_URL.scheme == "https":
            conn = http_client.HTTPSConnection(PARSED_BASE_URL.netloc)
        else:
            conn = http_client.HTTPConnection(PARSED_BASE_URL.netloc)
        url = f"{API_BASE_URL}{endpoint}"
        logging.info(
            f"querying: {method} {PARSED_BASE_URL.scheme}://{PARSED_BASE_URL.netloc}{url}"
        )
        logging.debug(f"query body: {body}")
        try:
            if body is not None:
                body = json.dumps(body)
            conn.request(method, url, body=body, headers=HEADERS)
        except Exception as e:
            raise RuntimeError(f"failed to query server: {e}")

        return conn.getresponse()


    def parse_response(response):
        try:
            tmp = response.read().decode("utf-8")
            logging.debug("gitea response: %s", tmp)
            return json.loads(tmp)
        except Exception as e:
            raise RuntimeError(f"failed to read server response: {e}")


    def do_get(endpoint):
        try:
            response = do_request("GET", endpoint)
        except Exception as e:
            raise RuntimeError(f"request failed: {e}")

        logging.info(f"query response status: {response.status}")

        if response.status == http.HTTPStatus.NOT_FOUND:
            return None

        if response.status != http.HTTPStatus.OK:
            raise RuntimeError(f"failed to GET: {response.reason}")

        return parse_response(response)


    def do_post(endpoint, data):
        try:
            response = do_request("POST", endpoint, data)
        except Exception as e:
            raise RuntimeError(f"request failed: {e}")

        logging.info(f"query response status: {response.status}")

        if response.status != http.HTTPStatus.CREATED:
            raise RuntimeError(f"failed to POST: {response.reason}")

        return parse_response(response)


    def do_put(endpoint, data):
        try:
            response = do_request("PUT", endpoint, data)
        except Exception as e:
            raise RuntimeError(f"request failed: {e}")

        logging.info(f"query response status: {response.status}")

        if response.status != http.HTTPStatus.NO_CONTENT:
            raise RuntimeError(f"failed to PUT: {response.reason}")

        return None


    def do_patch(endpoint, data):
        try:
            response = do_request("PATCH", endpoint, data)
        except Exception as e:
            raise RuntimeError(f"request failed: {e}")

        logging.info(f"query response status: {response.status}")

        if response.status != http.HTTPStatus.OK:
            raise RuntimeError(f"failed to PATCH: {response.reason}")

        return parse_response(response)


    def do_delete(endpoint):
        try:
            response = do_request("DELETE", endpoint)
        except Exception as e:
            raise RuntimeError(f"request failed: {e}")

        logging.info(f"query response status: {response.status}")

        if response.status == http.HTTPStatus.NOT_FOUND:
            return None

        if response.status != http.HTTPStatus.NO_CONTENT:
            raise RuntimeError(f"failed to DELETE: {response.reason}")

        return None


    #
    # repository stuff
    #
    def get_organization_repositories(organization):
        endpoint = f"/orgs/{organization}/repos"
        return do_get(endpoint)


    def create_organization_repository(organization, data):
        endpoint = f"/orgs/{organization}/repos"
        return do_post(endpoint, data)


    def delete_repository(owner, name):
        # raise NotImplementedError("not implemented: gitea api is strange")
        endpoint = f"/users/{owner}/repos/{name}"
        return do_delete(endpoint)


    def sync_repository(parent, _, delete=False):
        spec = parent["spec"]
        name = spec["name"]
        organization = spec["organization"]
        all_repositories = get_organization_repositories(organization)
        if all_repositories is None:
            return create_result(
                "error", message=f"organization '{organization}' does not exist"
            )

        existing = list(filter(lambda r: r["name"] == name, all_repositories))
        if len(existing) < 1:
            if delete:
                return create_result("deleted")

            try:
                existing = create_organization_repository(organization, spec)
            except Exception as e:
                raise RuntimeError(f"failed to create repository: {e}")

        else:
            existing = existing[0]

        if delete:
            owner = existing["owner"]
            # doesn't work if owner is org
            owner_name = owner["username"]
            # doesn't work either then
            # owner_name = ADMIN_USERNAME

            # just pretend ...
            try:
                delete_repository(owner_name, name)
                return create_result("deleted")
            except Exception as e:
                raise RuntimeError(f"failed to delete repository: {e}")

        return create_result("ready")


    #
    # organization stuff
    #
    def get_organization(name):
        endpoint = f"/orgs/{name}"
        return do_get(endpoint)


    def get_organization_teams(name):
        endpoint = f"/orgs/{name}/teams"
        return do_get(endpoint)


    def create_organization(data):
        endpoint = "/orgs"
        return do_post(endpoint, data)


    def delete_organization(name):
        endpoint = f"/orgs/{name}"
        return do_delete(endpoint)


    # TODO: don't hardcode ;)
    TEAM_UNITS = [
        "repo.pulls",
        "repo.projects",
        "repo.wiki",
        "repo.ext_wiki",
        "repo.packages",
        "repo.actions",
        "repo.code",
        "repo.issues",
        "repo.ext_issues",
        "repo.releases",
    ]

    TEAM_UNITS_MAP = {
        "repo.actions": "none",
        "repo.code": "none",
        "repo.ext_issues": "none",
        "repo.ext_wiki": "none",
        "repo.issues": "none",
        "repo.packages": "none",
        "repo.projects": "none",
        "repo.pulls": "none",
        "repo.releases": "none",
        "repo.wiki": "none",
    }

    team_preset = {
        "can_create_org_repo": False,
        "includes_all_repositories": False,
        "permission": "read",
        "units": TEAM_UNITS,
        "units_map": TEAM_UNITS_MAP,
    }


    def create_organization_team(name, data):
        endpoint = f"/orgs/{name}/teams"
        team = copy.deepcopy(team_preset)
        team.update(data)
        return do_post(endpoint, team)


    def patch_organization_team(team_id, data):
        endpoint = f"/teams/{team_id}"
        team = copy.deepcopy(team_preset)
        team.update(data)
        return do_patch(endpoint, team)


    def delete_organization_team(team_id):
        endpoint = f"/teams/{team_id}"
        return do_delete(endpoint)


    def sync_organization(parent, children, delete=False):
        spec = parent["spec"]
        username = spec["username"]
        if delete:
            try:
                delete_organization(username)
                return create_result("deleted")
            except Exception as e:
                raise RuntimeError(f"failed to delete organization: {e}")
        existing = get_organization(username)
        teams = []
        if existing is None:
            try:
                existing = create_organization(spec)
            except Exception as e:
                raise RuntimeError(f"failed to create org: {e}")
        else:
            try:
                teams = get_organization_teams(username)
            except Exception as e:
                raise RuntimeError(f"failed to get org teams: {e}")

        existing_teams_by_name = dict([(t["name"], t) for t in teams])
        to_delete = list(existing_teams_by_name.keys())
        for team in spec.get("teams", []):
            team_name = team["name"]
            if team_name not in existing_teams_by_name:
                try:
                    create_organization_team(username, team)
                except Exception as e:
                    raise RuntimeError(f"failed to create org team: {e}")
            else:
                to_delete.remove(team_name)
                existing_team = existing_teams_by_name[team_name]
                patched = copy.deepcopy(existing_team)
                patched.update(team)
                if patched != existing_team:
                    try:
                        patch_organization_team(existing_team["id"], patched)
                    except Exception as e:
                        raise RuntimeError(f"failed to patch org team: {e}")

        for team_name in to_delete:
            existing_team = existing_teams_by_name[team_name]
            try:
                delete_organization_team(existing_team["id"])
            except Exception as e:
                raise RuntimeError(f"failed to delete org team: {e}")

        return create_result("ready")


    #
    # user stuff
    #
    def get_user(name):
        endpoint = f"/users/{name}"
        return do_get(endpoint)


    def create_user(data):
        endpoint = "/admin/users"
        return do_post(endpoint, data)


    def patch_user(name, data):
        endpoint = f"/admin/users/{name}"
        return do_patch(endpoint, data)


    def delete_user(name):
        endpoint = f"/admin/users/{name}"
        return do_delete(endpoint)


    def create_pw():
        return "".join(secrets.choice(PASSWORD_APLHABET) for i in range(PASSWORD_LENGTH))


    def get_users_teams(username):
        endpoint = f"/user/teams?sudo={username}"
        return do_get(endpoint)


    def remove_users_team(team_id, username):
        endpoint = f"/teams/{team_id}/members/{username}"
        return do_delete(endpoint)


    def add_users_team(team_id, username):
        endpoint = f"/teams/{team_id}/members/{username}"
        return do_put(endpoint, None)


    def sync_user(parent, children, delete=False):
        spec = parent["spec"]
        username = spec["username"]
        if delete:
            try:
                delete_user(username)
                return create_result("deleted")
            except Exception as e:
                raise RuntimeError(f"failed to delete user: {e}")

        secret_name = parent["metadata"]["name"] + "-password"
        secrets = children.get("Secret.v1", {})
        secret = secrets.get(secret_name)
        password = None
        if secret is not None:
            password_b64 = secret["data"]["password"]
            password = from_b64string(password_b64)
        else:
            password = create_pw()
            password_b64 = b64string(password)

        existing = get_user(username)
        if existing is None:
            spec["password"] = password
            spec["login_name"] = spec.get("login_name", username)
            try:
                create_user(spec)
            except Exception as e:
                raise RuntimeError(f"failed to create user: {e}")
        else:
            try:
                patch_user(
                    username,
                    {
                        "username": username,
                        "password": password,
                        "login_name": existing.get("login_name", username),
                        "source_id": existing.get("source_id", 0),
                    },
                )
            except Exception as e:
                raise RuntimeError(f"failed to patch user: {e}")

        children = [
            {
                "apiVersion": "v1",
                "kind": "Secret",
                "type": "Opaque",
                "metadata": {"name": secret_name},
                "data": {
                    "password": password_b64,
                },
            }
        ]

        try:
            users_teams = get_users_teams(username)
        except Exception as e:
            raise RuntimeError(f"failed to user's teams: {e}")

        assigned_teams_lookup = dict(
            [((t["name"], t["organization"]["name"]), t) for t in users_teams]
        )
        to_remove = list(assigned_teams_lookup.keys())

        teams = spec.get("teams", [])
        for team in teams:
            team_name = team["name"]
            organization_name = team["organization"]
            key = (team_name, organization_name)
            if key in assigned_teams_lookup:
                to_remove.remove(key)
                continue

            try:
                organization_teams = get_organization_teams(organization_name)
            except Exception as e:
                raise RuntimeError(f"failed to get teams for organization: {e}")

            if not organization_teams:
                raise ValueError("organization has no teams")

            organization_team = list(
                filter(lambda t: t["name"] == team_name, organization_teams)
            )
            if len(organization_team) < 1:
                raise ValueError("team not found in organization")
            organization_team = organization_team[0]

            try:
                add_users_team(organization_team["id"], username)
            except Exception as e:
                raise RuntimeError(f"failed to add user to team: {e}")

        for key in to_remove:
            existing_team = assigned_teams_lookup[key]
            try:
                remove_users_team(existing_team["id"], username)
            except Exception as e:
                raise RuntimeError(f"failed to remove user from team: {e}")

        return create_result("ready", children=children)


    #
    # application stuff
    #
    def get_applications():
        endpoint = "/user/applications/oauth2"
        return do_get(endpoint)


    def get_application(application_id):
        endpoint = f"/user/applications/oauth2/{application_id}"
        return do_get(endpoint)


    def create_application(data):
        endpoint = "/user/applications/oauth2"
        return do_post(endpoint, data)


    def patch_application(application_id, data):
        endpoint = f"/user/applications/oauth2/{application_id}"
        return do_patch(endpoint, data)


    def delete_application(application_id):
        endpoint = f"/user/applications/oauth2/{application_id}"
        return do_delete(endpoint)


    def sync_application(parent, children, delete=False):
        spec = parent["spec"]
        name = spec["name"]
        if delete:
            try:
                delete_application(name)
                return create_result("deleted")
            except Exception as e:
                raise RuntimeError(f"failed to delete application: {e}")

        secret_name = parent["metadata"]["name"] + "-secret"
        secrets = children.get("Secret.v1", {})
        secret = secrets.get(secret_name)
        client_secret_b64 = client_id_b64 = ""
        if secret is not None:
            client_secret_b64 = secret["data"]["clientSecret"]
            client_secret = from_b64string(client_secret_b64)
            client_id_b64 = secret["data"]["clientId"]
            client_id = from_b64string(client_id_b64)

        all_applications = get_applications()
        existing = list(filter(lambda a: a["name"] == name, all_applications))
        if len(existing) < 1:
            if delete:
                return create_result("deleted")

            try:
                app = create_application(spec)
            except Exception as e:
                raise RuntimeError(f"failed to create application: {e}")

            client_id = app["client_id"]
            client_id_b64 = b64string(client_id)
            client_secret = app["client_secret"]
            client_secret_b64 = b64string(client_secret)
        else:
            existing = existing[0]
            existing_id = existing["id"]
            if delete:
                try:
                    delete_application(existing_id)
                    return create_result("deleted")
                except Exception as e:
                    raise RuntimeError(f"failed to delete application: {e}")

            if client_secret_b64 == "" or client_id_b64 == "":
                try:
                    app = patch_application(existing_id, spec)
                except Exception as e:
                    raise RuntimeError(f"failed to create application: {e}")

                client_id = app["client_id"]
                client_id_b64 = b64string(client_id)
                client_secret = app["client_secret"]
                client_secret_b64 = b64string(client_secret)

        children = [
            {
                "apiVersion": "v1",
                "kind": "Secret",
                "type": "Opaque",
                "metadata": {"name": parent["metadata"]["name"] + "-secret"},
                "data": {
                    "clientId": client_id_b64,
                    "clientSecret": client_secret_b64,
                },
            }
        ]

        return create_result("ready", children=children)


    #
    # token stuff
    #
    def get_user_tokens(username):
        endpoint = f"/users/{username}/tokens"
        return do_get(endpoint)


    def create_user_token(username, data):
        endpoint = f"/users/{username}/tokens"
        return do_post(endpoint, data)


    def delete_user_token(username, name):
        endpoint = f"/users/{username}/tokens/{name}"
        return do_delete(endpoint)


    def sync_token(parent, children, delete=False):
        spec = parent["spec"]
        username = spec["username"]
        all_tokens = get_user_tokens(username)
        if all_tokens is None:
            raise RuntimeError("user does not exist")

        name = spec["name"]
        if delete:
            try:
                delete_user_token(username, name)
                return create_result("deleted")
            except Exception as e:
                raise RuntimeError(f"failed to delete user token: {e}")

        token = token_b64 = ""
        secret_name = parent["metadata"]["name"] + "-secret"
        secrets = children.get("Secret.v1", {})
        secret = secrets.get(secret_name)
        if secret is not None:
            token_b64 = secret["data"]["token"]
            token = from_b64string(token_b64)

        existing = list(filter(lambda t: t["name"] == name, all_tokens))
        if len(existing) == 0:
            try:
                existing = create_user_token(username, spec)
            except Exception as e:
                raise RuntimeError(f"failed to create user token: {e}")
        else:
            existing = existing[0]
            if token_b64 == "" or set(existing["scopes"]) != set(spec["scopes"]):
                try:
                    delete_user_token(username, name)
                except Exception as e:
                    raise RuntimeError(f"failed to delete user token for update: {e}")

                try:
                    existing = create_user_token(username, spec)
                except Exception as e:
                    raise RuntimeError(f"failed to re-create user token: {e}")
                token = existing["sha1"]
                token_b64 = b64string(token)

        assert token_b64 != ""
        children = [
            {
                "apiVersion": "v1",
                "kind": "Secret",
                "type": "Opaque",
                "metadata": {"name": parent["metadata"]["name"] + "-secret"},
                "data": {
                    "token": token_b64,
                },
            }
        ]
        return create_result("ready", children=children)


    class Dispatcher(http_server.BaseHTTPRequestHandler):
        def do_POST(self):
            handler = globals().get(f"sync_{self.path[1:]}")
            if handler is None:
                raise RuntimeError(f"no handler found for {self.path[1:]}")

            content_length = int(self.headers["content-length"])
            observed = json.loads(self.rfile.read(content_length))
            logging.debug(f"received: {observed}")
            result = create_result("error", message="something didn't work")
            delete = observed.get("finalizing", False)
            parent = observed["parent"]
            parent["spec"] = recase(parent["spec"])
            try:
                result = handler(
                    parent,
                    observed.get("children", {}),
                    delete=delete,
                )
            except Exception as e:
                logging.exception("sync exception")
                result = create_result("error", message=f"sync failed: {type(e)}: {e}")
            else:
                if delete:
                    result["finalized"] = True

            logging.debug(f"result:\n{pprint.pformat(result)}")

            self.send_response(http.HTTPStatus.OK)
            self.send_header("Content-type", "application/json")
            self.end_headers()
            self.wfile.write(json.dumps(result).encode())

        def log_error(self, fmt, *args):
            logging.error(fmt, *args)

        def log_message(self, fmt, *args):
            logging.info(fmt, *args)


    if __name__ == "__main__":
        log_format = "%(asctime)s %(levelname)s %(name)s: %(message)s"
        log_level = os.getenv("LOG_LEVEL", "INFO")
        log_cfg = {
            "version": 1,
            "formatters": {"standard": {"format": log_format}},
            "disable_existing_loggers": True,
            "handlers": {
                "default": {
                    "class": "logging.StreamHandler",
                    "formatter": "standard",
                    "stream": "ext://sys.stdout",
                    "level": log_level,
                }
            },
            "loggers": {
                "": {"handlers": ["default"], "level": log_level},
            },
        }

        import logging.config

        logging.config.dictConfig(log_cfg)
        bind_host, bind_port = "", 8080
        logging.info(f"starting http server on {bind_host}:{bind_port}")
        server = http_server.HTTPServer((bind_host, bind_port), Dispatcher)
        server.serve_forever()
kind: ConfigMap
metadata:
  creationTimestamp: null
  name: gitea-resource-controller
  namespace: gitea