access.py

#!/usr/bin/env python3

import webbrowser
import urllib
import requests
import re
import subprocess
import tempfile
import os
import shlex
import sys

# Get URL from user and extract the CI namespace and CI cluster used for this job
url = input("Please enter your Prow Job URL: ")
*_, job, test_id = urllib.parse.urlparse(url).path.split("/")

# This part won't work if the job's already done
logs_url = f"https://prow.ci.openshift.org/log?container=test&id={test_id}&job={job}"
print(f"Checking {logs_url}")
resp = requests.get(logs_url)
try:
    namespace_url = (
        re.compile(r"Using namespace(.*)", re.MULTILINE).findall(resp.text)[0].strip()
    )
except IndexError:
    print("Failed to find CI Console URL from job logs, is it possible that job is already finished? This script is intended for live jobs that are still running")
    sys.exit(1)

# namespace_url looks something like https://console-openshift-console.apps.build03.ky4t.p1.openshiftapps.com/k8s/cluster/projects/ci-op-pw10vvhv
*_, namespace = urllib.parse.urlparse(namespace_url).path.split("/")
netloc = urllib.parse.urlparse(namespace_url).netloc
ci_cluster = (
    re.compile(r"console-openshift-console.apps.(.*).openshiftapps.com", re.MULTILINE)
    .findall(netloc)[0]
    .strip()
)

# Direct user to browser to get `oc login` command
input(f"Press any key to open your browser, then please log in to the OpenShift console, and copy the oc login command")
token_request = (
    f"https://oauth-openshift.apps.{ci_cluster}.openshiftapps.com/oauth/token/request"
)
webbrowser.open(token_request)
oc_login_command = input(
    "Please paste the oc login command you copied from your browser: "
)

# Login to the CI cluster with the user provided `oc login` command,
# extract KUBECONFIG for the cluster that we're installing during this
with tempfile.NamedTemporaryFile(delete=False) as ci_kubeconfig:
    print(f"Creating temporary CI cluster kubeconfig at {ci_kubeconfig.name}")
    env = {**os.environ, "KUBECONFIG": ci_kubeconfig.name}

    # Login to the job project
    print(f"Attempting to log in using {oc_login_command}...")
    subprocess.run(shlex.split(oc_login_command), env=env)
    print(f"Setting oc project to {namespace}")
    subprocess.run(["oc", "project", namespace], env=env)

# Save kubeconfig from secret into temporary file, give the user the shell export command to start using it
with tempfile.NamedTemporaryFile(delete=False) as cluster_kubeconfig:
    print("To login to the installer/test-runner cluster, run this command: ")
    print(f"export KUBECONFIG={ci_kubeconfig.name}")