Improve estimation, error handling, verification, etc.

Author: aegilops

estimate_push_protection_rate.py

@@ -19,6 +19,12 @@ def add_args(parser: argparse.ArgumentParser) -> None:
         type=str,
         help="Path to the file containing the list of patterns with push protection",
     )
+    parser.add_argument(
+        "--cut-off-date",
+        type=str,
+        default=None,
+        help="ISO date string to filter secrets detected after this date (e.g., 2023-01-01)",
+    )
 
 
 def main() -> None:
@@ -35,33 +41,85 @@ def main() -> None:
     with open(args.secrets_file, "r") as f:
         secrets = json.load(f)
 
-    total_secrets = len(secrets)
+    secrets_count = len(secrets)
     protected_secrets = [secret for secret in secrets if secret.get("secret_type") in patterns]
+    protected_secrets_count = len(protected_secrets)
 
-    print(f"Total secrets: {total_secrets}")
-    print(f"Protected secrets: {len(protected_secrets)}")
+    print(f"Total secrets: {secrets_count}")
+    print(f"Protected secrets: {protected_secrets_count}")
 
-    if total_secrets > 0:
-        protection_rate = (len(protected_secrets) / total_secrets) * 100
+    if secrets_count > 0:
+        protection_rate = (protected_secrets_count / secrets_count) * 100
         print(f"Estimated push protection rate: {protection_rate:.2f}%")
     else:
         print("No secrets found to evaluate.")
+        return
 
     # now evaluate how often we'd expect to block pushes, using the `first_commit_date` field
     # that's in ISO format with a Z suffix
     now = datetime.now(timezone.utc)
 
-    # find the oldest blocked commit
-    earliest_blocked_commit_date = min([
-        datetime.fromisoformat(secret["first_commit_date"].replace("Z", "+00:00"))
-        for secret in protected_secrets
-    ])
+    cut_off_date = args.cut_off_date
+    cut_off_datetime = None
+
+    if cut_off_date is not None:
+        try:
+            # add a time and TZ if just a date is provided
+            if len(cut_off_date) == 10:
+                cut_off_date += "T00:00:00+00:00"
+            # Handle 'Z' suffix for UTC
+            if cut_off_date.endswith("Z"):
+                cut_off_date = cut_off_date.replace("Z", "+00:00")
+            cut_off_datetime = datetime.fromisoformat(cut_off_date)
+            remaining_protected_secrets = [
+                secret for secret in protected_secrets
+                if "first_commit_date" in secret and datetime.fromisoformat(secret["first_commit_date"].replace("Z", "+00:00")) >= cut_off_datetime
+            ]
+            remaining_secrets = [
+                secret for secret in secrets
+                if "first_commit_date" in secret and datetime.fromisoformat(secret["first_commit_date"].replace("Z", "+00:00")) >= cut_off_datetime
+            ]
+        except ValueError:
+            print(f"Invalid cut-off date format: {cut_off_date}. Expected ISO format.")
+            return
+        
+        if not remaining_protected_secrets:
+            print("No protected secrets found after applying cut-off date filter.")
+            return
+        else:
+            remaining_secrets_count = len(remaining_secrets)
+            remaining_protected_secrets_count = len(remaining_protected_secrets)
+            print(f"Total secrets after cut-off date: {remaining_secrets_count}")
+            print(f"Protected secrets after cut-off date: {remaining_protected_secrets_count}")
+            protection_rate = (remaining_protected_secrets_count / remaining_secrets_count) * 100
+            print(f"Estimated push protection rate after cut-off date: {protection_rate:.2f}%")
+    else:
+        remaining_protected_secrets = protected_secrets
 
-    blocking_timespan = now - earliest_blocked_commit_date
-    rate = len(protected_secrets) / blocking_timespan.days if blocking_timespan.days > 0 else len(protected_secrets)
+    # get FPs for closed secrets, and estimate for any open ones
+    false_positives = 0
+
+    false_positives += sum([1 for secret in remaining_protected_secrets if secret.get("state") == "closed" and secret.get("resolution") == "false_positive"])
+    false_positives += sum([1 for secret in remaining_protected_secrets if secret.get("state") == "open"]) // 100
+
+    print(f"Measured + expected false positives: {false_positives}")
+
+    if cut_off_date:
+        earliest_date = cut_off_datetime
+    else:
+    # find the oldest blocked commit with an accessible commit
+        earliest_date = min((
+            datetime.fromisoformat(secret["first_commit_date"].replace("Z", "+00:00")) if "first_commit_date" in secret else now
+            for secret in remaining_protected_secrets
+        ))
 
-    print(f"Estimated secrets blocked per day since {earliest_blocked_commit_date.date()}: {rate:.2f}")
+    blocking_timespan = now - earliest_date
+    rate = len(remaining_protected_secrets) / blocking_timespan.days if blocking_timespan.days > 0 else len(remaining_protected_secrets)
 
+    print(f"Estimated secrets blocked per day since {earliest_date.date()}: {rate:.2f}")
+    print(f"                      ... per week ...            : {rate * 7:.2f}")
+    print(f"                      ... per month ...           : {rate * 30:.2f}")
+    print(f"                      ... per year ...            : {rate * 365:.2f}")
 
 if __name__ == "__main__":
     main()

githubapi.py

@@ -50,12 +50,13 @@ class RateLimited(Exception):
 class GitHub:
     """A GitHub API client."""
 
-    def __init__(self, token: str | None = None, hostname="github.com") -> None:
+    def __init__(self, token: str | None = None, hostname="github.com", verify: bool | str = True) -> None:
         token = token if token is not None else os.getenv("GITHUB_TOKEN")
         if token is None:
             raise ValueError("GITHUB_TOKEN environment variable must be set")
 
         self.session = requests.Session()
+        self.session.verify = verify
         self.session.headers.update({"Authorization": f"Bearer {token}"})
         self.session.headers.update({"Accept": "application/vnd.github.v3+json"})
         self.session.headers.update({"X-GitHub-Api-Version": "2022-11-28"})
@@ -304,7 +305,7 @@ def paginate(
             )
 
         if progress:
-            pbar = tqdm(desc="GitHub API", unit=" requests")
+            pbar = tqdm(desc="Paging with GitHub API", unit="page")
             pbar.reset(total=None)
 
         direction = ""
@@ -430,6 +431,7 @@ def list_secret_scanning_alerts(
         scope: str = "org",
         bypassed: bool = False,
         generic: bool = False,
+        progress: bool = True,
     ) -> Generator[dict, None, None]:
         """List secret scanning alerts for a GitHub repository, organization or Enterprise."""
         query = {"state": state} if state is not None else {}
@@ -445,6 +447,7 @@ def list_secret_scanning_alerts(
             since=since,
             date_field="created_at",
             paging="cursor",
+            progress=progress,
         )
 
         results = (