aides-spec/aides_spec/commands/checksums.py

import hashlib
import os
import subprocess

import requests
import typer

app = typer.Typer()


def calculate_sha256(file_path):
    sha256_hash = hashlib.sha256()
    with open(file_path, "rb") as f:
        for byte_block in iter(lambda: f.read(4096), b""):
            sha256_hash.update(byte_block)
    return sha256_hash.hexdigest()


def run_bash_command(command):
    result = subprocess.run(
        command,
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
    )
    if result.returncode != 0:
        raise RuntimeError(f"Ошибка при выполнении команды: {result.stderr}")
    return result.stdout.strip()


def download_file(url, dest_folder="downloads"):
    os.makedirs(dest_folder, exist_ok=True)
    local_filename = os.path.join(dest_folder, os.path.basename(url))
    with requests.get(url, stream=True, timeout=3000) as r:
        r.raise_for_status()
        with open(local_filename, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)
    return local_filename


def update_checksums(script_path):
    # Выполняем bash-скрипт, чтобы получить массив sources
    command = (
        f"source {script_path} && printf '%s\\n' \"${{sources_amd64[@]}}\""
    )
    try:
        sources_output = run_bash_command(command)
    except RuntimeError as e:
        print(f"Ошибка выполнения скрипта: {e}")
        return

    sources = sources_output.splitlines()
    if not sources:
        print("Не удалось получить массив sources_amd64.")
        return

    new_checksums = []
    for source_url in sources:
        print(f"Скачивание: {source_url}")
        local_file = download_file(source_url)
        checksum = calculate_sha256(local_file)
        new_checksums.append(f"sha256:{checksum}")
        os.remove(local_file)

    new_checksums_block = " ".join(f"'{chk}'" for chk in new_checksums)
    print(new_checksums_block)


@app.command(help="Get checksums for new sources")
def checksums():
    update_checksums(f"{os.getcwd()}/alr.sh")