def get_possible_credentials()

in github-runner-ami/packer/files/runner-supervisor.py [0:0]


def get_possible_credentials(repo: str) -> List[str]:
    client = boto3.client("ssm")
    paginator = client.get_paginator("describe_parameters")

    path = os.path.join('/runners/', repo, '')
    baked_path = os.path.join(path, 'runnersList')

    # Pre-compute the list, to avoid making lots of requests and getting throttled by SSM API in case of
    # thundering herd
    try:
        log.info("Using pre-computed credentials indexes from %s", baked_path)
        resp = client.get_parameter(Name=baked_path)
        return resp['Parameter']['Value'].split(',')
    except client.exceptions.ParameterNotFound:
        pass

    log.info("Looking at %s for possible credentials", path)

    pages = paginator.paginate(
        ParameterFilters=[{"Key": "Path", "Option": "Recursive", "Values": [path]}],
        PaginationConfig={
            "PageSize": 50,
        },
    )

    seen = set()

    for i, page in enumerate(pages):
        log.info("Page %d", i)
        for param in page['Parameters']:
            name = param['Name']
            log.info("%s", name)

            # '/runners/x/1/config' -> '1/config',
            # '/runners/x/y/1/config' -> 'y/1/config',
            local_name = name[len(path) :]

            try:
                # '1/config' -> '1'
                index, _ = local_name.split('/')
            except ValueError:
                # Ignore any 'x/y' when we asked for 'x'. There should only be an index and a filename
                log.debug("Ignoring nested path %s", name)
                continue

            try:
                # Check it's a number, but keep variable as string
                int(index)
            except ValueError:
                log.debug("Ignoring non-numeric index %s", name)
                continue

            index = os.path.basename(os.path.dirname(name))
            seen.add(index)

    if not seen:
        raise RuntimeError(f'No credentials found in SSM ParameterStore for {repo!r}')

    try:
        resp = client.put_parameter(
            Name=baked_path, Type='StringList', Value=','.join(list(seen)), Overwrite=False
        )
        log.info("Stored pre-computed credentials indexes at %s", baked_path)
    except client.exceptions.ParameterAlreadyExists:
        # Race, we lost, never mind!
        pass

    return list(seen)