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()