Skip to content

Generic Package Checksum

With the current approach the package is downloaded directly after an upload to compare the sha256sum.

Gitlab v 14.5.X provides an attribute select for generic paackages, that can be used to get a json response for the upload: https://docs.gitlab.com/ee/user/packages/generic_packages/#publish-a-package-file

It's also possible to use the gitlab API, but this currently requires a private token instead a job token. Here an approach that failed for this reason:

import requests
import sys

warningArgs = """\
This script requires exactly 5 arguments

Args:   gitlabAccessToken   Token with gitlab API read scope
        pipelineId          Id of the current pipeline
                            Example: '106109'
        packageEndpoint     Gitlab project package API endpoint
                            Example: https://gitlab.hzdr.de/api/v4/projects/3259/packages/
        packageName         Name of the generic package
                            Example: 'static'
        sha256sum           sha256sum to compare with
"""

if len(sys.argv) != 6:
    print (warningArgs)
    sys.exit("Script was called with " + str(len(sys.argv)-1) + " args" )

gitlabAccessToken = sys.argv[1]
pipelineId = sys.argv[2]
packageEndpoint = sys.argv[3]
packageName = sys.argv[4]
sha256sum = sys.argv[5]

# gitlabAccessToken = "mystery"
# pipelineId = 106109
# packageEndpoint = 'https://gitlab.hzdr.de/api/v4/projects/3259/packages/'
# packageName = 'static' 

packageQuery = {'package_name' : packageName}
# Use CI Access Token
# packageHeaders = {'PRIVATE-TOKEN' : 'mystery'}
packageHeaders = {'PRIVATE-TOKEN' : gitlabAccessToken}

# 
def findJSONObjectInPaginatedAPI(condition, url, params, headers, compareObjects = False):
    """
    Finds an JSON Object in an array of JSON objects. The array can be fetched via an API endpoint

    Parameters:
       condition        (function): The condition that the object should satisfy
       url              (str):      The API endpoint, that supports pagination
       params           (dict):     The parameter for the API endpoint
       headers          (dict):     The header for the API endpoint
       compareObjects   (Bool):     Parameter to find the most fitting Object, enable object comparison 
                                    within the condition and search the full list

    Returns:
       object(dict): python dict object which is compliant to the json object of the request
    """
    currentPage = 1
    totalPages = 1
    #

    # This condition could also be a while(true). The condition is cosmetic to
    # show how often the loop runs maximal
    currentObject = {}
    while (currentPage <= totalPages):
        # overwrite the page header
        params['page'] = currentPage
        response = requests.get(url, params=params, headers=headers)

        if response.status_code!= 200:
            print('Request failed with status code:' + str(response.status_code))
            # TODO Exit and show more info about request
        # Get the total pages for the first request only
        if (currentPage == 1) :
            # Parse to int to throw an error if the API is malfunctioning
            print(response.headers)
            print(response.json)
            print(url)
            print(params)
            # print(headers)
            totalPages = int(response.headers['x-total-pages'])
            # # Make sure total page is a fixed number
            # if (type(totalPages)) != int:
            #     print(type(totalPages))
            #     sys.exit('Total pages for request "'+ url + '" are not given as integer.')

        objectList = response.json()
        for object in objectList:
            if (compareObjects):
                if condition(object, currentObject):
                    currentObject = object
            else:
                if condition(object):
                    return object

        # Return object if any was found. Otherwise throw an error
        if currentPage == totalPages:
            if bool(currentObject):
                return currentObject
            sys.exit('Object not found for request "'+ url + '"')
        
        # increment the page for the next loop
        currentPage += 1

def packageCondition(package):
    if package['package_type'] == 'generic':
        return True
    return False

package = findJSONObjectInPaginatedAPI(packageCondition, packageEndpoint, packageQuery, packageHeaders)
packageId = package['id']

def filesCondition(currentFileMetaInfo, oldFileMetaInfo):
    # The json object with the file meta data contains a parameter pipelines,
    # which is an array with exact one entry: the pipline within the file was
    # uploaded.
    # Though the jobs of the pipeline can be triggered multiple times we check
    # the latest upload, determined by the highest id.
    # It is unlikely (not designed within git-lab) to run the same job of the
    # same pipeline, so the file with the highest id is the target.
    if currentFileMetaInfo['pipelines'][0]['id'] == pipelineId:
        if not bool(oldFileMetaInfo) or (bool(oldFileMetaInfo) and oldFileMetaInfo['id'] < currentFileMetaInfo['id']):
            return True
    return False

filesQuery = {'package_name' : packageName}
# Use CI Access Token
# filesHeaders = {'PRIVATE-TOKEN' : 'mystery'}
filesHeaders = {'PRIVATE-TOKEN' : gitlabAccessToken}

fileMetaJSON = findJSONObjectInPaginatedAPI(filesCondition, packageEndpoint + str(packageId) + '/package_files', filesQuery, filesHeaders, True)
print(fileMetaJSON['file_sha256'])

# TODO: Return Error if not matching to given sha256