Add Remediation Commit Support

.github/workflows/test.yml

@@ -42,20 +42,12 @@ jobs:
           virtualenvs-create: true
           virtualenvs-in-project: true
 
-      - name: Load cached venv
-        id: cached-poetry-dependencies
-        uses: actions/cache@5a3ec84eff668545956fd18022155c47e93e2684 # v4.2.0
-        with:
-          path: .venv
-          key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('**/poetry.lock') }}
-
       - name: Install dependencies
-        if: steps.cached-poetry-dependencies.outputs.cache-hit != 'true'
         run: poetry install --no-interaction
 
       - name: Run tests
         run: |
-          poetry run coverage run -m pytest tests.py
+          poetry run coverage run -m pytest tests/
           poetry run coverage xml
 
       - name: SonarCloud Scan

.gitignore

@@ -8,6 +8,17 @@ __pycache__/
 *.py[cod]
 *.egg-info/
 *.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
 build/
 .eggs/
 
@@ -25,3 +36,35 @@ htmlcov/
 
 # OS
 .DS_Store
+.AppleDouble
+.LSOverride
+
+# Icon must end with two \r
+Icon
+
+# Thumbnails
+._*
+
+# Files that might appear in the root of a volume
+.DocumentRevisions-V100
+.fseventsd
+.Spotlight-V100
+.TemporaryItems
+.Trashes
+.VolumeIcon.icns
+.com.apple.timemachine.donotpresent
+
+# Directories potentially created on remote AFP share
+.AppleDB
+.AppleDesktop
+Network Trash Folder
+Temporary Items
+.apdisk
+
+## Specific to this project
+
+/*.yml
+/*.csv
+/dco-signoffs/
+/remediation-commits/
+/*.log

contrib_check/commit.py

@@ -8,39 +8,87 @@
 # Provides a decorator-esque pattern on the GitPython.Commit object to add the commit inspection capabilities
 
 import re
+import logging
 
+# third party modules
+import yaml
 import git
 
 class Commit():
+
     # GitPython.Commit object
     git_commit_object = None
-
     repo_object = None
     is_merge_commit = False
 
     create_prior_commits_dir = 'dco-signoffs'
-
+
+    remediation_regex_individual = r"I,\s+(.*?)\s+<(.*?)>,\s+hereby\s+add\s+my\s+Signed-off-by\s+to\s+this\s+commit:\s+([a-f0-9]+)"
+    remediation_regex_thirdparty = r"On\s+behalf\s+of\s+(.*?)\s+<(.*?)>,\s+I,\s+(.*?)\s+<(.*?)>,\s+hereby\s+add\s+my\s+Signed-off-by\s+to\s+this\s+commit:\s+([a-f0-9]+)"
+
+    allow_remediation_commit_individual = False
+    allow_remediation_commit_thirdparty = False
+
+    remediations = []
+
     def __init__(self, git_commit_object, repo_object):
         self.git_commit_object = git_commit_object
         self.repo_object = repo_object
         self.is_merge_commit = len(git_commit_object.parents) > 1
-
-    def checkDCOSignoff(self):
-        if self.isDCOSignOffRequired():
-            return self.hasDCOSignOff() or self.hasDCOPastSignoff()
-
+
+        self.load_remediation_commit_config()
+
+    def check_dco_signoff(self):
+        if self.is_dco_signoff_required():
+            return self.has_dco_signoff() or self.has_dco_past_signoff() or self.has_remediation()
+
         return True
 
-    def isDCOSignOffRequired(self):
+    def is_dco_signoff_required(self):
         return not self.is_merge_commit
-
-    def hasDCOSignOff(self):
-        return re.search("Signed-off-by: (.+)",self.git_commit_object.message)
 
-    def hasDCOPastSignoff(self):
+    def has_dco_signoff(self):
+        return (re.search("Signed-off-by: (.+)",self.git_commit_object.message) != None)
+
+    def has_dco_past_signoff(self):
         for signoff in self.repo_object.past_signoffs:
-            if re.search(self.git_commit_object.hexsha.encode(),signoff):
+            if (re.search(self.git_commit_object.hexsha.encode(),signoff) != None):
                 return True
 
-        return False;
+        return False
+
+    def has_remediation(self):
+        return self.repo_object.git_repo_object.git.rev_parse(self.git_commit_object.hexsha, short="7") in self.remediations
+
+    def load_remediation_commit_config(self):
+        try:
+            with open(self.repo_object.git_repo_object.head.commit.tree[".github/dco.yml"].abspath, 'r') as file:
+                config = yaml.safe_load(file)
+                self.allow_remediation_commit_individual = config['allowRemediationCommits']['individual'] if config and 'allowRemediationCommits' in config and 'individual' in config['allowRemediationCommits'] else False
+                self.allow_remediation_commit_thirdparty = config['allowRemediationCommits']['thirdParty'] if config and 'allowRemediationCommits' in config and 'thirdParty' in config['allowRemediationCommits'] else False
+        except KeyError:
+            return False
+        else:
+            return True
+
+    def is_remediation_commit(self):
+        is_remediation_commit = False
+
+        if self.allow_remediation_commit_individual:
+            logging.getLogger().debug(f"Looking for individual remediation commits for commit {self.git_commit_object.hexsha}")
+            for match in re.findall(self.remediation_regex_individual,self.git_commit_object.message,flags=re.I|re.M|re.DOTALL):
+                # ensure it's a valid remediation commit by matching the author with the attestation
+                if ( match[0] == self.git_commit_object.author.name ) and ( match[1] == self.git_commit_object.author.email ):
+                    logging.getLogger().debug(f"Found individual remediation commit {match[2]} in commit {self.git_commit_object.hexsha}")
+                    self.remediations.append(match[2])
+                    is_remediation_commit = True
+        if self.allow_remediation_commit_thirdparty:
+            logging.getLogger().debug(f"Looking for third party remediation commits for commit {self.git_commit_object.hexsha}")
+            for match in re.findall(self.remediation_regex_thirdparty,self.git_commit_object.message,flags=re.I|re.M|re.DOTALL):
+                # ensure it's a valid remediation commit by matching the author with the attestation
+                if ( match[2] == self.git_commit_object.author.name ) and ( match[3] == self.git_commit_object.author.email ):
+                    logging.getLogger().debug(f"Found third party remediation commit {match[4]} in commit {self.git_commit_object.hexsha}")
+                    self.remediations.append(match[4])
+                    is_remediation_commit = True
 
+        return is_remediation_commit

contrib_check/main.py

@@ -8,13 +8,14 @@
 import shutil
 from argparse import ArgumentParser, FileType
 from datetime import datetime
+import logging
+import sys
 
 # third party modules
 import yaml
 from contrib_check.repo import Repo
 from contrib_check.org import Org
 
-
 def main():
     startTime = datetime.now()
 
@@ -24,9 +25,39 @@ def main():
     group.add_argument("--repo", dest="repo", help="URL or path to the repo to search")
     group.add_argument("--org", dest="org", help="URL to GitHub org to search")
     parser.add_argument("--dco", dest="dco", help="Perform a DCO check (defaults to true)", default=True)
+    parser.add_argument("--dco-allow-individual-remediation-commits",
+                        dest="dco-allow-individual-remediation-commits",
+                        help="Allow individual remediation commits for DCO signoffs (only needed if not enabled in dco.yml in the repo)",
+                        default=False)
+    parser.add_argument("--dco-allow-thirdparty-remediation-commits",
+                        dest="dco-allow-thirdparty-remediation-commits",
+                        help="Allow third party remediation commits for DCO signoffs (only needed if not enabled in dco.yml in the repo)",
+                        default=False)
+    parser.add_argument("--dco-allow-dcosignoffs",
+                        dest="dco-allow-dcosignoffs",
+                        help="Allow reading and writing legacy DCO Signoff files")
+    parser.add_argument("-l", "--log", dest="loglevel", default="error",
+                        choices=['debug', 'info', 'warning', 'error', 'critical'], help="logging level")
+    parser.add_argument("--logfile", default='debug.log', help="Name for the log file")
 
     args = parser.parse_args()
 
+    levels = {
+        'critical': logging.CRITICAL,   # errors that mean an immediate stop
+        'error': logging.ERROR,         # general errors that will effect the output
+        'warn': logging.WARNING,        # errors that can be caught and corrected
+        'warning': logging.WARNING,
+        'info': logging.INFO,           # infomational messages
+        'debug': logging.DEBUG          # messages to help debug things misbehaving ;-)
+    }
+    handlers = [logging.FileHandler(args.logfile,mode="w")]
+    handlers.append(logging.StreamHandler(sys.stdout))
+    logging.basicConfig(
+        level=levels.get(args.loglevel.lower()),
+        format="%(asctime)s [%(levelname)s] %(message)s",
+        handlers=handlers
+    )
+
     config = {}
     if args.configfile:
         config = yaml.safe_load(args.configfile)
@@ -38,11 +69,11 @@ def main():
         if 'repo' in config:
             args.repo = config['repo']
         elif 'org' in config:
-            args.org = config['org']
+            args.org = config['org']['name']
 
     repos = []
     if args.org:
-        orgObj = Org(args.org)
+        orgObj = Org(args.org, load_repos=False)
         if 'type' in config['org']:
             orgObj.org_type = config['org']['type']
         if 'ignore_repos' in config['org']:
@@ -51,24 +82,20 @@ def main():
             orgObj.only_repos = config['org']['only_repos']
         if 'skip_archived' in config['org']:
             orgObj.skip_archived = config['org']['skip_archived']
-        repos = orgObj.loadRepos()
+        repos = orgObj.reload_repos()
 
     if args.repo:
         repos = [Repo(args.repo)]
 
     for repoObj in repos:
-        print("Searching repo {}...".format(repoObj.name))
+        logging.getLogger().info(f"Searching repo {repoObj.name}")
         if 'dco' in config or args.dco:
             if 'dco' in config and 'prior_commits_directory' in config['dco']:
                 repoObj.prior_commits_dir = config['dco']['prior_commits_directory']
             if 'dco' in config and 'signoff_dirs' in config['dco']:
-                repoObj.loadPastSignoffs(config['dco']['signoff_dirs'])
+                repoObj.load_past_signoffs(config['dco']['signoff_dirs'])
             else:
-                repoObj.loadPastSignoffs()
+                repoObj.load_past_signoffs()
             repoObj.scan()
 
-    print("This took " + str(datetime.now() - startTime))
-
-
-if __name__ == '__main__':
-    main()
+    logging.getLogger().info("This took {} seconds".format(str(datetime.now() - startTime)))