script which allows to connect to a tmate session on github CI.

github_ci_ssh.py

"""
This script serves to connect to a tmate session via SSH on a github CI runner in order to
enable interactive debugging.

Status: This script is an experimental version, neither documented nor well tested and
contains obsolete code. However, it is useful for the author and might be so for others.

Motivation: On the CI platform software sometimes behaves differently then on local machines.
The usual debugging procedure is lengthy because for every change (e.g. adding debug output via
print()) a new commit and a new CI run has to take place. Opening an SSH connection to the
runner can thus drastically simplify CI debugging. Such a connection can be opened by adding
the following step to the yaml file for the workflow:

```
- name: Setup tmate session
    uses: mxschmitt/action-tmate@v3
```

This results in printing the SSH-connection details to the interactive output of the CI job
from where it can be copied to a local terminal.


This script allows to omit copy-pasting these details by fechting the required information via
the github API. This requires some addintional steps on the CI runner, see the following yaml code
taken from (https://github.com/cknoll/xaiev/blob/main/.github/workflows/python-app.yml):


```
    # ...
    
    # temporary debugging (uncomment if necessary)
    # - name: Setup tmate session
    #   id: tmate
    #   uses: mxschmitt/action-tmate@v3
    #   with:
    #     detached: true

    # # this exports the ad-hoc ssh-connection address to an text file such that it can be accessed via API
    # # this allows to connect to the debugging ssh server without manual copy-pasting

    # - name: Capture tmate output
    #   run: |
    #       echo $(cat /home/runner/work/_temp/_runner_file_commands/* | grep ::notice::SSH | awk '{print $NF}') > tmate-connection.txt

    # - name: Upload tmate logs as artifact
    #   uses: actions/upload-artifact@v4
    #   with:
    #     name: tmate-connection
    #     path: tmate-connection.txt
```

"""

import os
import requests
import re
import subprocess
from dotenv import load_dotenv

import zipfile
import io

# optional, for debugging
from ipydex import IPS


load_dotenv("./.env_dev")


REPO_OWNER = os.getenv("REPO_OWNER")
REPO_NAME = os.getenv("REPO_NAME")

WORKFLOW_ID = os.getenv("WORKFLOW_ID")

GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")


# GitHub repository information


def get_latest_run_id():
    url = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/actions/workflows/{WORKFLOW_ID}/runs"
    headers = {
        "Authorization": f"token {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json"
    }
    params = {
        "status": "in_progress",
        "per_page": 1
    }

    response = requests.get(url, headers=headers, params=params)
    response.raise_for_status()

    runs = response.json().get("workflow_runs", [])
    return runs[0]["id"] if runs else None

def get_html(run_id):
    url = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/actions/runs/{run_id}/jobs"
    headers = {
        "Authorization": f"token {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json"
    }

    response = requests.get(url, headers=headers)
    response.raise_for_status()
    response_dict = response.json()

    # IPS()
    html_url = response_dict["jobs"][0]["html_url"]

    response2 = requests.get(html_url, headers=headers)
    response2.raise_for_status()
    html_content = response2.content.decode("utf8")
    IPS()

    # response2_dict = response2.json()

    exit()

def get_job_id(run_id):
    url = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/actions/runs/{run_id}/jobs"
    headers = {
        "Authorization": f"token {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json"
    }

    response = requests.get(url, headers=headers)
    response.raise_for_status()
    response_dict = response.json()

    # IPS()
    return response_dict["jobs"][0]["id"]

def get_jobs(run_id):
    jobs_url = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/actions/runs/{run_id}/jobs"
    headers = {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json"
    }
    response = requests.get(jobs_url, headers=headers)
    jobs = response.json()["jobs"]
    IPS()

headers = {
        "Authorization": f"token {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json"
    }


def get_artifacts(run_id):

    artifacts_url = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/actions/runs/{run_id}/artifacts"
    response = requests.get(artifacts_url, headers=headers)

    if response.status_code != 200:
        print(f"Failed to fetch artifacts: {response.status_code}")
        print(response.json())
        exit(1)

    artifacts = response.json()['artifacts']
    if not artifacts:
        print("no artifacts found -> ssh shell not yet created")
        exit(1)
    artifact_url = artifacts[0]["archive_download_url"]
    r2 = requests.get(artifact_url, headers=headers, stream=True)
    zip_file = io.BytesIO(r2.content)

    with zipfile.ZipFile(io.BytesIO(r2.content)) as zip_file:
        # Assuming there is only one text file in the zip archive
        # You can also specify the exact file name if you know it
        for file_name in zip_file.namelist():
            with zip_file.open(file_name) as text_file:
                content = text_file.read().decode('utf-8')  # Read and decode the content
                break  # Stop after reading the first text file

    return content


def get_ssh_command(job_id):
    # job_id = "37387571458"
    url = f"https://api.github.com/repos/{REPO_OWNER}/{REPO_NAME}/actions/jobs/{job_id}/logs"
    headers = {
        "Authorization": f"token {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json"
    }

    response = requests.get(url, headers=headers)
    response.raise_for_status()

    logs = response.text

    IPS()
    ssh_command_match = re.search(r'ssh \S+@\S+\.tmate\.io', logs)
    return ssh_command_match.group(0) if ssh_command_match else None


def main():
    run_id = get_latest_run_id()
    if not run_id:
        print("No active workflow run found.")
        return


    ssh_connection = get_artifacts(run_id)

    ssh_command = f"ssh {ssh_connection}"
    print("Connecting to CI worker...")
    subprocess.run(ssh_command, shell=True)

    exit()

    # obsolete attempts
    # get_html(run_id)
    # get_jobs(run_id)
    # job_id = get_job_id(run_id)

    # ssh_command = get_ssh_command(job_id)
    # if not ssh_command:
    #     print("SSH command not found in the logs.")
    #     return

    # print("Connecting to CI worker...")
    # subprocess.run(ssh_command, shell=True)

if __name__ == "__main__":
    main()