| #!/usr/bin/env {{ python }} |
| VERSION = '2.0' |
| |
| import io |
| import os |
| import re |
| import subprocess |
| import sys |
| |
| if sys.stdout.isatty() and os.path.exists('/dev/tty'): |
| TTY = io.TextIOWrapper(io.FileIO(os.open('/dev/tty', os.O_NOCTTY | os.O_RDWR), 'r+')) |
| else: |
| TTY = None |
| |
| TOPLEVEL = subprocess.check_output(['git', 'rev-parse', '--show-toplevel'], encoding='utf-8').strip() |
| |
| LOCATION = os.path.join(TOPLEVEL, r'{{ location }}') |
| SCRIPTS = os.path.dirname(os.path.dirname(LOCATION)) |
| sys.path.append(SCRIPTS) |
| |
| REMOTE_RE = re.compile(r'{{ remote_re }}') |
| NULL_HASH = '0' * 40 |
| TRUNCATED_COMMIT_LEN = 12 |
| |
| CLASS_1 = 1 # Commit exists on another remote |
| CLASS_2 = 2 # Commit is a cherry-pick of one that exists on another remote |
| CLASS_3 = 3 # Commit references a bug which requires redaction |
| |
| COMMIT_REF_BASE = r'r?R?[a-f0-9A-F]+(\.\d+)?@?([0-9a-zA-z\-\/\.]+[0-9a-zA-z\-\/])?' |
| COMPOUND_COMMIT_REF = r'(?P<primary>{})(?P<secondary> \({}\))?'.format(COMMIT_REF_BASE, COMMIT_REF_BASE) |
| CHERRY_PICK_RE = [ |
| re.compile(r'\S* ?[Cc]herry[- ][Pp]ick of {}'.format(COMPOUND_COMMIT_REF)), |
| re.compile(r'\S* ?[Cc]herry[- ][Pp]ick {}'.format(COMPOUND_COMMIT_REF)), |
| re.compile(r'\S* ?[Cc]herry[- ][Pp]icked {}'.format(COMPOUND_COMMIT_REF)), |
| ] |
| UNPACK_SECONDARY_RE = re.compile(r' \(({})\)'.format(COMMIT_REF_BASE)) |
| |
| DEV_BRANCHES = re.compile(r'.*[(eng)(dev)(bug)]/.+') |
| PROD_BRANCHES = re.compile(r'\S+-[\d+\.]+-branch') |
| DEFAULT_BRANCHES = ('master', 'main') |
| |
| QUIET = -1 |
| VERBOSE = 1 |
| VERBOSITY = int(os.environ.get('VERBOSITY', '0')) |
| |
| SECURITY_LEVELS = {{ security_levels }} |
| MAX_LEVEL = max(SECURITY_LEVELS.values(), default=0) |
| |
| |
| def print_error(why): |
| if sys.stdout.isatty(): |
| RED_TEXT_FORMAT = '\033[31m{}\033[0m\n' |
| (TTY or sys.stderr).write(RED_TEXT_FORMAT.format(why)) |
| (TTY or sys.stderr).flush() |
| else: |
| sys.stderr.write('{}\n'.format(why)) |
| |
| |
| def print_tty(message): |
| (TTY or sys.stdout).write('{}\n'.format(message)) |
| (TTY or sys.stdout).flush() |
| |
| |
| DEFAULT_MODE = 'default' # Forbid class 1, prompt for class 2 and 3 |
| PUBLISH_MODE = 'publish' # Prompt for class 1, forbid class 2 and 3 |
| NO_RADAR_MODE = 'no-radar' # Forbid class 1 and 2, don't compute class 3 |
| MODE = os.environ.get('PUSH_HOOK_MODE', {{ default_pre_push_mode }}) |
| |
| |
| if MODE not in (DEFAULT_MODE, PUBLISH_MODE, NO_RADAR_MODE): |
| print_error("'{}' is not a recognized push mode".format(MODE)) |
| sys.exit(1) |
| |
| |
| REPOSITORY = None |
| try: |
| import webkitpy |
| from webkitcorepy import Version |
| from webkitscmpy import local, Commit |
| from webkitbugspy import version, Tracker |
| |
| if MODE != NO_RADAR_MODE and version >= Version(0, 9, 8): |
| REPOSITORY = local.Git(os.getcwd()) |
| |
| except: |
| pass |
| |
| if not REPOSITORY: |
| print_error('Checkout too out-of-date to check bug status of novel commits') |
| |
| |
| def security_level_for_remote(remote): |
| match = REMOTE_RE.match(remote) |
| if not match: |
| return None |
| return SECURITY_LEVELS.get('{}:{}'.format(match.group('host'), match.group('path')).lower(), None) |
| |
| |
| def security_level_for_remotes(remotes, name_to_remote=None): |
| if not name_to_remote: |
| _, name_to_remote = remote_name_mappings() |
| security_levels = [security_level_for_remote(name_to_remote.get(name, name)) for name in remotes] |
| if 0 in security_levels: |
| return 0 |
| if None in security_levels: |
| return None |
| return min(security_levels) |
| |
| |
| def commits_in(from_ref, to_ref): |
| command = ['git', 'rev-list'] |
| if from_ref: |
| command.append('{}..{}'.format(from_ref, to_ref)) |
| else: |
| command.append(to_ref) |
| proc = None |
| try: |
| proc = subprocess.Popen( |
| command, |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE, |
| encoding='utf-8', |
| ) |
| line = proc.stdout.readline() |
| while line: |
| yield line.rstrip()[:TRUNCATED_COMMIT_LEN] |
| line = proc.stdout.readline() |
| finally: |
| if proc and proc.poll() is None: |
| proc.kill() |
| |
| |
| def remote_name_mappings(): |
| a, b = {}, {} |
| for line in subprocess.check_output( |
| ['git', 'remote', '-v'], |
| encoding='utf-8', |
| ).splitlines(): |
| remote = line.split() |
| if len(remote) < 2 or remote[1] == 'INVALID': |
| continue |
| match = REMOTE_RE.match(remote[1]) |
| if match: |
| b[remote[0]] = remote[1] |
| a['{}:{}'.format(match.group('host'), match.group('path'))] = remote[0] |
| return a, b |
| |
| |
| def remotes_for(commit): |
| if not commit: |
| return None |
| try: |
| result = set() |
| for line in subprocess.check_output( |
| ['git', 'branch', '-r', '--contains', commit], |
| stderr=subprocess.STDOUT, |
| encoding='utf-8', |
| ).splitlines(): |
| result.add(line.lstrip().split('/')[0]) |
| return list(result) |
| except subprocess.CalledProcessError: |
| return None |
| |
| |
| def branch_for(commit): |
| if not commit: |
| return None |
| try: |
| candidate = None |
| for line in subprocess.check_output( |
| ['git', 'branch', '-a', '--contains', commit], |
| stderr=subprocess.STDOUT, |
| encoding='utf-8', |
| ).splitlines(): |
| if line[0] == '*': |
| line = line[1:] |
| line = line.lstrip().rstrip() |
| is_remote = line.startswith('remotes') |
| if is_remote: |
| line = line.split('/', 2)[-1] |
| if not candidate: |
| candidate = line |
| continue |
| if candidate in DEFAULT_BRANCHES or PROD_BRANCHES.match(candidate): |
| continue |
| if PROD_BRANCHES.match(line) or DEV_BRANCHES.match(candidate): |
| candidate = line |
| |
| return candidate |
| except subprocess.CalledProcessError: |
| return None |
| |
| |
| def message_for(commit): |
| if not commit: |
| return None |
| return subprocess.check_output( |
| ['git', 'log', '--format=%B', '-n', '1', commit], |
| encoding='utf-8', |
| ) |
| |
| |
| def security_level_for_commit(commit, remotes, name_to_remote=None, allow_class_1=False): |
| if not name_to_remote: |
| _, name_to_remote = remote_name_mappings() |
| |
| # Class 1: the commit we're pushing exists on a remote already |
| class_1_level = None |
| if remotes: |
| class_1_level = security_level_for_remotes(remotes, name_to_remote) |
| # The commit is public already, we should preform no further checks |
| if class_1_level == 0: |
| return class_1_level, CLASS_1 |
| # The commit exists on another remote, and we are not allowing class 1 violations |
| if not allow_class_1 and remotes: |
| return class_1_level, CLASS_1 |
| |
| message = message_for(commit) |
| title = message.splitlines()[0] |
| |
| # Class 2: The commit we're pushing is a cherry-pick of something that exists on a remote already |
| class_2_level = None |
| for r in CHERRY_PICK_RE: |
| match = r.match(title) |
| if not match: |
| continue |
| primary = match.group('primary') |
| secondary = match.group('secondary') |
| if secondary: |
| secondary = UNPACK_SECONDARY_RE.match(secondary).groups()[0] |
| for candidate in filter(bool, (primary, secondary)): |
| original_remotes = remotes_for(candidate) |
| if not original_remotes: |
| continue |
| class_2_level = security_level_for_remotes(original_remotes, name_to_remote) |
| |
| # If the commit is a cherry-pick of something already public, the state of the bug it's referencing doesn't matter |
| if class_2_level == 0: |
| return class_2_level, CLASS_3 |
| # Check for a class 3 violation before returning our class 2 violation |
| break |
| break |
| |
| # Class 3: We must inspect the commit for bug references |
| if REPOSITORY: |
| did_redact = False |
| obj = Commit(hash=commit, message=message) |
| for issue in obj.issues: |
| redaction = issue.redacted |
| if redaction: |
| did_redact = True |
| if getattr(redaction, 'exemption', False): |
| return 0, CLASS_3 |
| if did_redact: |
| return 1, CLASS_3 |
| |
| if class_1_level is not None: |
| return class_1_level, CLASS_1 |
| if class_2_level is not None: |
| return class_2_level, CLASS_2 |
| |
| return None, None |
| |
| |
| def main(name, remote): |
| target_security_level = security_level_for_remote(remote) |
| if VERBOSITY > QUIET: |
| if target_security_level: |
| print_tty('Pushing to {}, which is classified as a secure remote'.format(name or remote)) |
| elif target_security_level == 0: |
| print_tty('Pushing to {}, which is classified as a public remote'.format(name or remote)) |
| else: |
| print_tty('Pushing to {}, which has no classification. Treating it as a public remote'.format(name or remote)) |
| |
| to_push = [] |
| for push in sys.stdin.readlines(): |
| _, local_ref, _, remote_ref = push.split() |
| # Deletes are always safe |
| if local_ref == NULL_HASH: |
| continue |
| if remote_ref == NULL_HASH: |
| remote_ref = None |
| to_push.append((remote_ref, local_ref)) |
| if not to_push: |
| if VERBOSITY > QUIET: |
| print_tty('No novel content being pushed') |
| return 0 |
| |
| remote_to_name, name_to_remote = remote_name_mappings() |
| if name == remote: |
| match = REMOTE_RE.match(remote) |
| if not match: |
| print_error("'{}' does not conform to a recognized target remote".format(remote)) |
| return 255 |
| name = remote_to_name.get('{}:{}'.format(match.group('host'), match.group('path')), remote) |
| if name not in name_to_remote: |
| print_error("Failed to map '{}' to an existing remote".format(remote)) |
| |
| remotes_with = {} |
| for from_ref, to_ref in to_push: |
| for commit in commits_in(from_ref, to_ref): |
| if commit not in remotes_with: |
| remotes_with[commit] = remotes_for(commit) or [] |
| |
| # Any future commits (being ancestors of this one) will nessesarily be on the same remote this commit is |
| rs = remotes_with.get(commit, []) |
| if not rs: |
| continue |
| if name in rs or MODE != PUBLISH_MODE and rs: |
| break |
| commit_level = security_level_for_remotes(rs, name_to_remote) |
| if commit_level == 0: |
| break |
| if (commit_level or MAX_LEVEL) <= (target_security_level or 0): |
| break |
| remotes_with = dict(filter(lambda pair: name not in pair[1], remotes_with.items())) |
| |
| print_tty("Verifying commits in '{}' mode".format(MODE)) |
| if VERBOSITY > QUIET: |
| print_tty('Attempting to push {} batch{} of commits to {}...'.format(len(to_push), '' if len(to_push) == 1 else 'es', name)) |
| |
| prompt_for = [] |
| for commit, remotes in remotes_with.items(): |
| security_level, categorization_class = security_level_for_commit(commit, remotes, name_to_remote=name_to_remote, allow_class_1=MODE == PUBLISH_MODE) |
| |
| # Unkown security level of source |
| if categorization_class == 1 and security_level is None and not target_security_level: |
| print_error("'{}' comes from an uncategorized remote, it's not safe to push it publically".format(commit)) |
| return 1 |
| # Security level of source exceeds security level of target |
| if categorization_class == 1 and (MAX_LEVEL if security_level is None else security_level) > (target_security_level or 0): |
| print_error("'{}' comes from a more secure remote than '{}'".format(commit, name)) |
| if MODE == PUBLISH_MODE: |
| prompt_for.append(commit) |
| else: |
| return 1 |
| |
| # Unknown security level from original of cherry-pick |
| if categorization_class == 2 and security_level is None and not target_security_level: |
| print_error("'{}' cherry-picks a change from an uncategorized remote, it's not safe to push it publically".format(commit)) |
| return 1 |
| # Security level of original of a cherry-pick exceeds security level of target |
| if categorization_class == 2 and (MAX_LEVEL if security_level is None else security_level) > (target_security_level or 0): |
| print_error("'{}' cherry-picks a change from a more secure remote than '{}'".format(commit, name)) |
| if MODE in (DEFAULT_MODE, PUBLISH_MODE): |
| prompt_for.append(commit) |
| else: |
| return 1 |
| |
| # Commit references a redacted issue, but we're pushing it publically |
| if categorization_class == 3 and (MAX_LEVEL if security_level is None else security_level) > (target_security_level or 0): |
| print_error("'{}' fixes a redacted issue, but '{}' is a public remote".format(commit, name)) |
| return 1 |
| |
| if VERBOSITY >= VERBOSE: |
| print_tty(' {} is security level {} by class {}'.format(commit, security_level, categorization_class)) |
| |
| if prompt_for and not TTY: |
| print_tty('No TTY available for prompting user, rejecting push') |
| return 1 |
| |
| if prompt_for: |
| branches = sorted({branch_for(commit) for commit in prompt_for}) |
| expected_response = 'publicize {}'.format( |
| branches[0] if len(branches) == 1 |
| else '{}, and {}'.format(', '.join(branches[:-1]), branches[-1]), |
| ) |
| print_tty('') |
| print_tty('We have detected you are about to publish security content to a public remote') |
| print_tty('Unless you are engaged in merge-back or fixing a security bug which has not shipped, you are likely making a mistake') |
| print_tty('') |
| sys.stdout.write('Type "{}" to continue with publication: '.format(expected_response)) |
| sys.stdout.flush() |
| response = TTY.readline().rstrip() |
| if response != expected_response: |
| if response: |
| print_error('Response does not match prompt') |
| print_error('User failed to explicitly approve potentially dangerous push') |
| return 1 |
| print_tty("User has explicitly approved potentially dangerous push, continuing...") |
| |
| elif VERBOSITY > QUIET: |
| print_tty('Verified {} novel commit{} to {} are safe to push'.format(len(remotes_with), '' if len(remotes_with) == 1 else 's', name)) |
| |
| return 0 |
| |
| if __name__ == '__main__': |
| try: |
| sys.exit(main(*sys.argv[1:])) |
| except KeyboardInterrupt: |
| print_tty('') |
| print_error('User has canceled push') |
| sys.exit(1) |
