"""

try:

spool += json.loads(json_string)

except json.decoder.JSONDecodeError as error:

spool += json.loads(json_string[:error.pos])

json_multiloads(json_string[error.pos:], spool)

def __init__(self, command, command_exit_code, command_stdout, command_stderr, http_status, http_headers):

self.command = command

self.command_exit_code = command_exit_code

self.command_stdout = command_stdout

self.command_stderr = command_stderr

self.http_status = http_status

self.http_headers = http_headers

def __init__(self, name, repository):

self.name = name

self.repository = repository

def __str__(self):

return self.name

def head(self):

"""Returns the SHA-1 commit ID of the HEAD."""

log.info('%s: Retrieving the HEAD SHA-1 commit ID of the branch "%s"', self.repository, self.name)

return self.repository.api(

f'repos/{self.repository}/commits/heads/{self.name}',

headers={'Accept': 'application/vnd.github.sha'},

raw_output=True,

)

def branch(self, new_branch_name):

"""Create a new branch from the HEAD of this branch."""

log.info('%s: Creating branch "%s"', self.repository, new_branch_name)

self.repository.api(

f'repos/{self.repository}/git/refs',

method='POST',

params={'ref': f'refs/heads/{new_branch_name}', 'sha': self.head()},

)

return GitHeadRef(new_branch_name, self.repository)

def create_file(self, name, content):

"""Create a file in this branch of the repository."""

log.info('%s: creating workflow file "%s"', self.repository, name)

self.repository.api(

f'repos/{self.repository}/contents/{name}',

method='PUT',

params={

'message': f'Create {name}',

'content': base64.b64encode(content.encode('utf-8')).decode('utf-8'),

'branch' : self.name,

},

)

def request_pull(self, title, body):

log.info('%s: creating Pull Request', self.repository)

params = {'title': title, 'head' : self.name, 'base' : self.repository.ref().name}

if body is not None:

params['body'] = body

# This regular expression captures the relevant bits from `gh api --include`. In particular:

# .group(1) = HTTP Status Code ("200")

# .group(2) = HTTP response headers

# .group(3) = HTTP response body

_gh_api_regex = re.compile(r'HTTP/\d\.\d (\d+) (?:\w[\w\s]+)\n(.+?)(?:\n|\r\n){2}([^\n]+)\n?', re.DOTALL)

def __init__(self, nwo, gh_exe):

self.gh_exe = gh_exe

self._object = self.api(f'repos/{nwo}')

@classmethod

def with_kwargs(cls, **kwargs):

"""

return lambda nwo: cls(nwo, **kwargs)

def __str__(self):

return self.name_with_owner

@property

def name_with_owner(self):

return self._object['full_name']

@property

def languages(self):

return self.api(f'repos/{self.name_with_owner}/languages')

def api(self, endpoint, method='GET', headers=None, params=None, raw_output=False):

command = [ self.gh_exe, 'api', endpoint, '--include', f'--method={method}' ]

if method == 'GET':

command.append('--paginate')

if headers is not None:

command.extend(f'--header={h}:{headers[h]}' for h in headers)

if isinstance(params, str):

command.append('--input=-')

elif isinstance(params, dict):

command.extend(f'--field={p}={params[p]}' for p in params)

params = None

for seconds in map(lambda x: x ** x, range(10)):

try:

process = subprocess.run(command, input=params, text=True, capture_output=True, check=True)

break

except subprocess.CalledProcessError as error:

log.debug('command=%s exit-status=%d\n%s', error.cmd, error.returncode, error.stdout)

http_status, http_headers, _ = self._gh_api_regex.findall(error.stdout)[0]

http_headers = { m[0]: m[1] for m in re.findall(r'([\w-]+):\s?(.+?)\r?\n', http_headers) }

# TODO: Use a case-insensitive dictionary/data structure.

# HTTP headers are case-insensitive. However, the way we specify them (as keys in

# a dictionary) is definitely case-sensitive. It creates the possibility for error.

# Check for primary rate limits. GitHub indicates a primary rate limit through the use of

# a 403 status and a set of particular headers.

# https://docs.github.com/en/rest/overview/resources-in-the-rest-api#checking-your-rate-limit-status

if http_status == '403' and http_headers['X-Ratelimit-Remaining'] == '0':

log.warning('GitHub primary rate limit reached; sleeping for %s seconds.', seconds)

time.sleep(int(seconds))

# GitHub may sometimes apply a secondary rate limit (to prevent abuse). Sometimes, it will tell

# us how to long to wait:

# >When you have been limited, use the Retry-After response header to slow down. The value of the

# >Retry-After header will always be an integer, representing the number of seconds you should wait

# >before making requests again.

# https://docs.github.com/en/rest/guides/best-practices-for-integrators#dealing-with-secondary-rate-limits

elif http_status == '403' and 'Retry-After' in http_headers:

log.warning('GitHub secondary rate limit reached; sleeping for %s seconds.', http_headers['Retry-After'])

time.sleep(int(http_headers['Retry-After']))

# GitHub may apply the secondary rate limit with little indication,

# other than a 403 status and an error message:

# >Requests that create content which triggers notifications, such as issues, comments and pull requests,

# >may be further limited and will not include a Retry-After header in the response.

# https://docs.github.com/en/rest/guides/best-practices-for-integrators#dealing-with-secondary-rate-limits

elif http_status == '403' and 'secondary rate limit' in error.stderr:

log.warning('GitHub secondary rate limit reached; sleeping for %d seconds.', seconds)

time.sleep(seconds)

else:

raise GithubError(

command=error.cmd,

command_exit_code=error.returncode,

command_stdout=error.stdout,

command_stderr=error.stderr,

http_status=int(http_status),

http_headers=http_headers,

) from error

log.debug('command=%s exit-status=%d\n%s', process.args, process.returncode, process.stdout)

buffer = io.StringIO()

# The output from `gh api --paginate --include` is a concatenation of individual `gh api --include`

# commands. Each one will have an HTTP line, headers, and a body to parse. So, we will use regular

# expressions to find each response in the output and handle them individually.

for match in self._gh_api_regex.finditer(process.stdout):

buffer.write(match.group(3))

buffer.seek(0)

if raw_output:

return buffer.read()

else:

return json.load(buffer)

def analyses(self, ref=None):

if ref is None:

params = None

else:

params = {'ref': ref}

analyses_json_raw = self.api(f'repos/{self.name_with_owner}/code-scanning/analyses', params=params, raw_output=True)

for analysis_json_obj in json_multiloads(analyses_json_raw):

yield CodeScanningAnalysis.from_api(self, analysis_json_obj, self.gh_exe)

def alerts(self):

alerts_json_raw = self.api(f'repos/{self.name_with_owner}/code-scanning/alerts', params={'per_page':100}, raw_output=True)

for alert_object in json_multiloads(alerts_json_raw):

yield CodeScanningAlert(alert_object)

def enable_advanced_security(self):

log.info('%s: enabling GitHub Advanced Security...', self.name_with_owner)

# Skip public repositories, which are enabled by default.

if self._object['visibility'] == 'public':

return

# Skip repositories that are already enabled.

if self._object['security_and_analysis']['advanced_security']['status'] == 'enabled':

return

self.api(

f'repos/{self.name_with_owner}',

method='PATCH',

params='{"security_and_analysis":{"advanced_security":{"status":"enabled"}}}',

)

def ref(self, name=None):

"""Returns a GitHeadRef object corresponding to the git branch of the repository, if one exists."""

def __init__(self, repository, analysis_id, gh_exe, ref=None, commit_sha=None, analysis_key=None, environment=None, category=None,

created_at=None, results_count=None, rules_count=None, sarif_id=None, deletable=False):

self.repository = repository

self.id = analysis_id

self.gh_exe = gh_exe

self.ref = ref

self.commit_sha = commit_sha

self.analysis_key = analysis_key

self.environment = environment

self.category = category

self.created_at = created_at

self.results_count = results_count

self.rules_count = rules_count

self.sarif_id = sarif_id

self.deletable = deletable

@classmethod

def from_api(cls, repository, obj, gh_exe):

return cls(

repository,

obj['id'],

gh_exe,

ref=obj['ref'],

commit_sha=obj['commit_sha'],

analysis_key=obj['analysis_key'],

environment=obj['environment'],

category=obj['category'],

created_at=obj['created_at'],

results_count=obj['results_count'],

rules_count=obj['rules_count'],

sarif_id=obj['sarif_id'],

deletable=obj['deletable'],

)

def is_deletable(self):

return self.deletable

def delete(self):

"""

next_analysis = self.repository.api(f'repos/{self.repository}/code-scanning/analyses/{self.id}?confirm_delete=true', method='DELETE')

if next_analysis['next_analysis_url'] is None and next_analysis['confirm_delete_url'] is None:

return None

next_analysis_id = int(next_analysis['next_analysis_url'].split('/')[-1])

def __init__(self, obj):

Rule = collections.namedtuple('Rule', ['name', 'id', 'severity', 'tags'])

Location = collections.namedtuple('Location', ['path', 'start_line', 'end_line', 'start_column', 'end_column'])

self.number = obj['number']

self.created_at = obj['created_at']

if obj['state'] in {'open', 'closed', 'fixed'}:

self.state = obj['state']

elif obj['dismissed_reason'] == 'false positive':

self.state = 'dismissed:false-positive'

elif obj['dismissed_reason'] == "won't fix":

self.state = 'dismissed:wont-fix'

elif obj['dismissed_reason'] == 'used in tests':

self.state = 'dismissed:used-in-tests'

else:

self.state = 'dismissed:other'

self.rule = Rule(obj['rule']['name'], obj['rule']['id'], obj['rule'].get('severity', obj['rule'].get('security_severity_level')), obj['rule']['tags'])

def normalize_lang(lang):

"""

if lang in {'C', 'C++'}:

return 'cpp'

if lang in {'Java', 'Kotlin'}:

return 'java'

if lang == 'C#':

return 'csharp'

if lang == 'Go':

return 'go'

if lang == 'Python':

return 'python'

if lang in {'JavaScript', 'TypeScript'}:

return 'javascript'

if lang == 'Ruby':

return 'ruby'

return None

random_weekly_cron = '{} {} {} {} {}'.format(

random.randint(0, 59), # minute

random.randint(0, 23), # hour

'*', # day of month

'*', # month

random.randint(0, 6) # day of week

)

branch_name = 'gh-code-scanning/codeql-analysis'

workflow_name = os.path.basename(args.workflow)

workflow_dstdir = '.github/workflows'

workflow_srcpath = os.path.abspath(args.workflow)

workflow_dstpath = os.path.join(workflow_dstdir, workflow_name)

if args.git_push and args.body is not None:

log.error('option -b, --body is incompatible with option --git-push, because option -b, --body implies the creation of a pull request.')

sys.exit(1)

with open(workflow_srcpath, encoding='utf-8') as workflow_file:

workflow_template = string.Template(workflow_file.read())

for repo in args.repos:

supported_langs = list(set(filter(None, map(normalize_lang, repo.languages))))

if len(supported_langs) == 0:

log.error('%s: no supported programming languages', repo)

continue

workflow_contents = workflow_template.safe_substitute(

GH_CS_DEFAULT_BRANCH_EXPR=repo.ref().name,

GH_CS_SCHEDULE_CRON_EXPR=repr(random_weekly_cron),

GH_CS_MATRIX_LANGUAGE_EXPR=repr(supported_langs)

)

if args.git_push:

branch = repo.ref()

else:

try:

branch = repo.ref().branch(branch_name)

except GithubError as error:

log.error('%s: cannot create branch "%s": %s', repo, branch_name, error)

continue

try:

branch.create_file(workflow_dstpath, workflow_contents)

except GithubError as error:

# TODO: if not args.git_push: branch.delete()

log.error('%s: cannot create file "%s": %s', repo, workflow_dstpath, error)

continue

if not args.git_push:

try:

branch.request_pull(args.message, args.body)

except GithubError as error:

# TODO: if not args.git_push: branch.delete()

log.error('%s: cannot create pull request: %s', repo, error)

continue

try:

repo.enable_advanced_security()

except GithubError as error:

log.warning('%s: cannot enable GitHub Advanced Security: %s', repo, error)

continue

for repo in args.repos:

for alert in repo.alerts():

print(repo,

format(alert.number, '>4d'),

alert.created_at,

format(alert.state, '>24s'),

alert.rule.id,

f'{alert.location.path}:{alert.location.start_line}',

flush=True,

if args.delete:

delete_queue = [ analysis for analysis in args.repo.analyses(ref=args.ref) if analysis.is_deletable() ]

while True:

try:

this_analysis = delete_queue.pop()

except IndexError:

return # done -- all analyses have been deleted!

next_analysis = this_analysis.delete()

if next_analysis is not None:

delete_queue.append(next_analysis)

print(this_analysis.id, flush=True)

# The "delete" flag (args.delete) was not passed, so print analyses instead of deleting them.

for analysis in args.repo.analyses(ref=args.ref):

print(args.repo,

format(analysis.id, '>9d'),

analysis.created_at,

analysis.analysis_key,

format(analysis.rules_count, '>3d'),

format(analysis.results_count, '4d'),

'Y' if analysis.is_deletable() else 'N',

analysis.sarif_id,

flush=True,

logging.basicConfig(format=f'%(asctime)s {os.getpid()} %(levelname)s %(message)s')

gh_exe = shutil.which('gh')

if gh_exe is None:

sys.exit('error: cannot find `gh` executable; please specify a path with `--gh`.')

parser = argparse.ArgumentParser()

parser.add_argument('--gh', metavar='path', default=gh_exe, help='path to the gh executable.')

parser.add_argument('--git', metavar='path', help='path to the git executable. [DEPRECATED]')

parser.add_argument('-v', '--verbose', action='count', default=0, help='print diagnostic information.')

parser.set_defaults(func=lambda x: parser.print_help())

subparsers = parser.add_subparsers()

parser_enable = subparsers.add_parser('enable', help='set up Code Scanning with GitHub CodeQL.')

parser_enable.add_argument('-f', '--force', action='store_true', help='overwrite existing workflow file.')

parser_enable.add_argument('--git-push', action='store_true', help='do not create PR; push commit to HEAD of default branch.')

parser_enable.add_argument('-m', '--message', default='Create CodeQL workflow file', help='specify the pull-request/commit message.')

parser_enable.add_argument('-b', '--body', help='specify the pull-request body.')

parser_enable.add_argument('-w', '--workflow', default=os.path.join(os.path.dirname(__file__), 'codeql-analysis.yml'), metavar='WORKFLOW_FILE_PATH', help='specify a custom CodeQL workflow file')

parser_enable.add_argument('repos', nargs='+', type=GithubRepository.with_kwargs(gh_exe=gh_exe))

parser_enable.set_defaults(func=do_enable)

parser_alerts = subparsers.add_parser('alerts', help='download code-scanning alerts.')

parser_alerts.add_argument('repos', nargs='+', type=GithubRepository.with_kwargs(gh_exe=gh_exe))

parser_alerts.set_defaults(func=do_alerts)

parser_analyses = subparsers.add_parser('analyses', help='list code-scanning analyses.')

parser_analyses.add_argument('-d', '--delete', action='store_true')

parser_analyses.add_argument('-r', '--ref')

parser_analyses.add_argument('repo', type=GithubRepository.with_kwargs(gh_exe=gh_exe))

parser_analyses.set_defaults(func=do_analyses)

args = parser.parse_args()

if args.git:

log.warning('option --git is deprecated; it will be unsupported in a future release.')

if args.verbose == 1:

log.setLevel(logging.INFO)

elif args.verbose > 1:

log.setLevel(logging.DEBUG)