diff --git a/sast/before.json b/sast/before.json new file mode 100644 index 0000000..c79d8e2 --- /dev/null +++ b/sast/before.json @@ -0,0 +1 @@ +{"version":"1.128.1","results":[],"errors":[{"code":2,"level":"error","type":"SemgrepError","message":"Invalid scanning root: repo"}],"paths":{"scanned":[]},"skipped_rules":[]} \ No newline at end of file diff --git a/src/autofic_core/app.py b/src/autofic_core/app.py index 4aa8064..9ded461 100644 --- a/src/autofic_core/app.py +++ b/src/autofic_core/app.py @@ -1,12 +1,13 @@ import os import sys -from pathlib import Path +import time import traceback - +from pathlib import Path from rich.console import Console + from autofic_core.errors import * -from autofic_core.utils.ui_utils import print_help_message -from autofic_core.pipeline import AutoFiCPipeline +from autofic_core.utils.ui_utils import print_help_message, print_divider +from autofic_core.pipeline import AutoFiCPipeline from autofic_core.log.log_writer import LogManager from autofic_core.log.log_generator import LogGenerator @@ -31,7 +32,7 @@ def __init__(self, explain, repo, save_dir, sast, llm, llm_retry, patch, pr): def run(self): try: - self._validate_options() + self.validate_options() if self.explain: print_help_message() return @@ -50,7 +51,7 @@ def run(self): pipeline.run() if self.pr: - self._run_pr(pipeline) + self.run_pr() except AutoficError as e: console.print(str(e), style="red") @@ -61,7 +62,7 @@ def run(self): console.print(traceback.format_exc(), style="red") sys.exit(1) - def _validate_options(self): + def validate_options(self): if not self.repo: raise NoRepositoryError() if not self.save_dir: @@ -75,55 +76,109 @@ def _validate_options(self): if not self.patch and self.pr: raise PRWithoutPatchError() - def _run_pr(self, pipeline): - # PR automation - branch_num = 1 + def run_pr(self): + try: + print_divider("PR Automation Stage") + + pr_procedure = self.initialize_pr_procedure() + + console.print("[1] Initializing PR process & checking branches...\n", style="bold cyan") + time.sleep(0.5) + pr_procedure.post_init() + pr_procedure.mv_workdir() + pr_procedure.check_branch_exists() + + console.print("\n[2] Notifying webhooks...\n", style="bold cyan") + time.sleep(0.5) + self.notify_webhooks(pr_procedure) + + console.print("\n[3] Creating and pushing PR workflow YAML...\n", style="bold cyan") + time.sleep(0.5) + self.handle_pr_yml(pr_procedure) + + console.print("\n[4] Changing files for the pull request...\n", style="bold cyan") + time.sleep(0.5) + pr_procedure.change_files() + + console.print("\n[5] Updating branch and creating Pull Request...\n", style="bold cyan") + time.sleep(0.5) + pr_procedure.current_main_branch() + pr_procedure.generate_pr() + pr_number = pr_procedure.create_pr() + + console.print(f"\n[ SUCCESS ] Pull Request created successfully!\n", style="bold green") + time.sleep(0.5) + + self.finalize_logging(pr_procedure, pr_number) + + except Exception as e: + console.print(f"[ PR ERROR ] {e}", style="bold red") + console.print(traceback.format_exc(), style="red") + raise + + def initialize_pr_procedure(self): base_branch = 'main' - branch_name = "UNKNOWN" - repo_name = "UNKOWN" - upstream_owner = "UNKOWN" save_dir = Path(self.save_dir).joinpath('repo') repo_url = self.repo.rstrip('/').replace('.git', '') - secret_discord = os.getenv('DISCORD_WEBHOOK_URL') - secret_slack = os.getenv('SLACK_WEBHOOK_URL') + json_path = str(Path(self.save_dir).joinpath("sast") / "before.json") token = os.getenv('GITHUB_TOKEN') user_name = os.getenv('USER_NAME') - - # Define PRProcedure class - json_path = str(Path(self.save_dir).joinpath("sast") / "before.json") tool = self.sast.lower() if self.sast else None - pr_procedure = PRProcedure( - base_branch, repo_name, upstream_owner, - save_dir, repo_url, token, user_name, json_path, tool + + return PRProcedure( + base_branch=base_branch, + repo_name="UNKNOWN", + upstream_owner="UNKNOWN", + save_dir=save_dir, + repo_url=repo_url, + token=token, + user_name=user_name, + json_path=json_path, + tool=tool ) - # Chapter 1 - pr_procedure.post_init() + + def notify_webhooks(self, pr_procedure): + secret_discord = os.getenv('DISCORD_WEBHOOK_URL') + secret_slack = os.getenv('SLACK_WEBHOOK_URL') + + user_name = pr_procedure.user_name repo_name = pr_procedure.repo_name - upstream_owner = pr_procedure.upstream_owner - # Chaper 2 - pr_procedure.mv_workdir() - # Chapter 3 - pr_procedure.check_branch_exists() - branch_name = pr_procedure.branch_name - # Chapter 4 + token = pr_procedure.token + EnvEncrypy(user_name, repo_name, token).webhook_secret_notifier('DISCORD_WEBHOOK_URL', secret_discord) EnvEncrypy(user_name, repo_name, token).webhook_secret_notifier('SLACK_WEBHOOK_URL', secret_slack) - # Chapter 5 - AboutYml().create_pr_yml() - AboutYml().push_pr_yml(user_name, repo_name, token, branch_name) - # Chapter 6 - pr_procedure.change_files() - # Chapter 7 - pr_procedure.current_main_branch() - # Chapter 8,9 - pr_procedure.generate_pr() - pr_number = pr_procedure.create_pr() - - # for log - repo_data = self.log_gen.generate_repo_log(save_dir=Path(self.save_dir), name=repo_name, owner=upstream_owner, - repo_url=repo_url, sastTool=tool, rerun=self.llm_retry) - pr_log_data = self.log_gen.generate_pr_log(owner=upstream_owner, repo=repo_name, user_name=user_name, - repo_url=repo_url, repo_hash=repo_data["repo_hash"], - pr_number=pr_number) + + def handle_pr_yml(self, pr_procedure): + user_name = pr_procedure.user_name + repo_name = pr_procedure.repo_name + token = pr_procedure.token + branch_name = pr_procedure.branch_name + + yml_handler = AboutYml() + yml_handler.create_pr_yml() + yml_handler.push_pr_yml(user_name, repo_name, token, branch_name) + + def finalize_logging(self, pr_procedure, pr_number): + tool = self.sast.lower() if self.sast else None + repo_url = self.repo.rstrip('/').replace('.git', '') + + repo_data = self.log_gen.generate_repo_log( + save_dir=Path(self.save_dir), + name=pr_procedure.repo_name, + owner=pr_procedure.upstream_owner, + repo_url=repo_url, + sastTool=tool, + rerun=self.llm_retry + ) + + pr_log_data = self.log_gen.generate_pr_log( + owner=pr_procedure.upstream_owner, + repo=pr_procedure.repo_name, + user_name=pr_procedure.user_name, + repo_url=repo_url, + repo_hash=repo_data["repo_hash"], + pr_number=pr_number + ) + self.log_manager.add_pr_log(pr_log_data) self.log_manager.add_repo_status(repo_data) \ No newline at end of file diff --git a/src/autofic_core/patch/apply_patch.py b/src/autofic_core/patch/apply_patch.py index dfec9f3..b03c9d2 100644 --- a/src/autofic_core/patch/apply_patch.py +++ b/src/autofic_core/patch/apply_patch.py @@ -50,7 +50,7 @@ def apply_all(self) -> bool: failed_patches.append(patch_file) if failed_patches: - self.console.print(f"\n[cyan][ INFO ] {len(failed_patches)} patches failed → trying overwrite from parsed (see logs)[/cyan]\n") + self.console.print(f"[cyan][ INFO ] {len(failed_patches)} patches failed → trying overwrite from parsed (see logs)[/cyan]\n") for patch_file in failed_patches: self.overwrite_with_parsed(patch_file) @@ -66,7 +66,7 @@ def apply_single(self, patch_file: Path) -> bool: ) if result.returncode == 0: - self.console.print(f"[white][✓] Patch applied: {patch_file.name}[/white]") + self.console.print(f"[white][✓] Patch applied: {patch_file.name}[/white]\n") return True else: self.console.print(PatchFailMessages.PATCH_FAILED.format(patch_file.name), style="yellow") @@ -161,7 +161,7 @@ def overwrite_with_parsed(self, patch_file: Path) -> bool: try: shutil.copyfile(matched_file, repo_file) - self.console.print(f"[white][✓] Overwrote repo file: {repo_file}[/white]") + self.console.print(f"[white][✓] Overwrote repo file: {repo_file}[/white]\n") return True except Exception as e: self.console.print(PatchErrorMessages.OVERWRITE_FAILED.format(e), style="red") diff --git a/src/autofic_core/pipeline.py b/src/autofic_core/pipeline.py index e5528fb..62a88a5 100644 --- a/src/autofic_core/pipeline.py +++ b/src/autofic_core/pipeline.py @@ -52,14 +52,14 @@ def clone(self): try: if self.handler.needs_fork: - console.print("\nAttempting to fork the repository...\n", style="cyan") + console.print("Attempting to fork the repository...\n", style="cyan") self.handler.fork() time.sleep(1) - console.print("\n[ SUCCESS ] Fork completed\n", style="green") + console.print("[ SUCCESS ] Fork completed\n", style="green") self.clone_path = Path( self.handler.clone_repo(save_dir=str(self.save_dir), use_forked=self.handler.needs_fork)) - console.print(f"\n[ SUCCESS ] Repository cloned successfully: {self.clone_path}\n", style="green") + console.print(f"[ SUCCESS ] Repository cloned successfully: {self.clone_path}", style="green") except ForkFailedError as e: sys.exit(1) @@ -77,8 +77,10 @@ def __init__(self, repo_path: Path, save_dir: Path): self.save_dir = save_dir def run(self): + description = "Running Semgrep...".ljust(28) with create_progress() as progress: - task = progress.add_task("[cyan]Running Semgrep...", total=100) + task = progress.add_task(description, total=100) + start = time.time() runner = SemgrepRunner(repo_path=str(self.repo_path), rule="p/javascript") result = runner.run_semgrep() @@ -123,8 +125,10 @@ def __init__(self, repo_path: Path, save_dir: Path): self.save_dir = save_dir def run(self): + description = "Running CodeQL...".ljust(28) with create_progress() as progress: - task = progress.add_task("[cyan]Running CodeQL...", total=100) + task = progress.add_task(description, total=100) + start = time.time() runner = CodeQLRunner(repo_path=str(self.repo_path)) result_path = runner.run_codeql() @@ -166,10 +170,12 @@ class SnykCodeHandler: def __init__(self, repo_path: Path, save_dir: Path): self.repo_path = repo_path self.save_dir = save_dir - + def run(self): + description = "Running SnykCode...".ljust(28) with create_progress() as progress: - task = progress.add_task("[cyan]Running SnykCode...", total=100) + task = progress.add_task(description, total=100) + start = time.time() runner = SnykCodeRunner(repo_path=str(self.repo_path)) result = runner.run_snykcode() @@ -267,7 +273,6 @@ def __init__(self, sast_result_path: Path, repo_path: Path, save_dir: Path, tool self.patch_dir = save_dir / "patch" def run(self): - console.print() print_divider("LLM Response Generation Stage") prompt_generator = PromptGenerator() @@ -284,10 +289,10 @@ def run(self): llm = LLMRunner() self.llm_output_dir.mkdir(parents=True, exist_ok=True) - - console.print("Starting GPT response generation\n") + + description = "Generating LLM responses... \n".ljust(28) with create_progress() as progress: - task = progress.add_task("[magenta]Generating LLM responses...", total=len(prompts)) + task = progress.add_task(description, total=len(prompts)) for p in prompts: response = llm.run(p.prompt) @@ -298,7 +303,7 @@ def run(self): time.sleep(0.01) progress.update(task, completed=100) - console.print(f"\n[ SUCCESS ] LLM responses saved → {self.llm_output_dir}\n", style="green") + console.print(f"[ SUCCESS ] LLM responses saved → {self.llm_output_dir}", style="green") return prompts, file_snippets def retry(self): @@ -370,7 +375,7 @@ def run(self): success = patch_applier.apply_all() if success: - console.print(f"\n[ SUCCESS ] All patches successfully applied\n", style="green") + console.print(f"[ SUCCESS ] All patches successfully applied", style="green") else: console.print(f"\n[ WARN ] Some patches failed to apply → {self.repo_dir}\n", style="yellow") diff --git a/src/autofic_core/pr_auto/create_yml.py b/src/autofic_core/pr_auto/create_yml.py index d2b595c..57e49c8 100644 --- a/src/autofic_core/pr_auto/create_yml.py +++ b/src/autofic_core/pr_auto/create_yml.py @@ -18,7 +18,9 @@ """ import os import subprocess -import click +from rich.console import Console + +console = Console() # Handles creation and git operations for GitHub Actions workflow YAML files class AboutYml: @@ -82,7 +84,7 @@ def push_pr_yml(self, user_name, repo_name, token, branch_name): """ repo_url = f'https://x-access-token:{token}@github.com/{user_name}/{repo_name}.git' subprocess.run(['git', 'remote', 'set-url', 'origin', repo_url], check=True) - click.secho("[ INFO ] Pushing the generated .github/workflows/pr_notify.yml.", fg="yellow") + console.print("[ INFO ] Created GitHub Actions workflow file: pr_notify.yml\n", style="white") subprocess.run(['git', 'add', '.github/workflows/pr_notify.yml'], check=True) - subprocess.run(['git', 'commit', '-m', "[Autofic] Create package.json and CI workflow"], check=True) - subprocess.run(['git', 'push', 'origin', branch_name], check=True) + subprocess.run(['git', 'commit', '-m', "[ AutoFiC ] Create package.json and CI workflow"], check=True) + subprocess.run(['git', 'push', 'origin', branch_name], check=True) \ No newline at end of file diff --git a/src/autofic_core/pr_auto/env_encrypt.py b/src/autofic_core/pr_auto/env_encrypt.py index a2c1573..7a74bee 100644 --- a/src/autofic_core/pr_auto/env_encrypt.py +++ b/src/autofic_core/pr_auto/env_encrypt.py @@ -14,69 +14,41 @@ # limitations under the License. # ============================================================================= -"""Contains their functional aliases. -""" - import base64 import requests from nacl import public, encoding +from rich.console import Console + +console = Console() class EnvEncrypy: - """ - Utility class for encrypting and registering GitHub Actions secrets (e.g., webhooks) - using the repository's public key. - """ def __init__(self, user_name, repo_name, token): - """ - Initialize with GitHub repository and user credentials. - - :param user_name: GitHub username (owner) - :param repo_name: GitHub repository name - :param token: GitHub personal access token - """ self.user_name = user_name self.repo_name = repo_name self.token = token def webhook_secret_notifier(self, secret_name: str, webhook_url: str): - """ - Registers a webhook URL as a secret in the target GitHub repository. - This fetches the repo's public key, encrypts the webhook URL, - and stores it as a new Actions secret. - - :param secret_name: The name of the secret to be created/updated - :param webhook_url: The actual webhook URL to store (as secret) - """ url = f'https://api.github.com/repos/{self.user_name}/{self.repo_name}/actions/secrets/public-key' headers = {'Authorization': f'token {self.token}'} resp = requests.get(url, headers=headers) pubkey_info = resp.json() - # Prevent KeyError (validates key info exists) if 'key_id' not in pubkey_info or 'key' not in pubkey_info: - print(f"[ERROR] Invalid public key info: {pubkey_info}") + console.print(f"[ ERROR ] Invalid public key info: {pubkey_info}", style="red") return key_id = pubkey_info['key_id'] encrypted_value = self.encrypt(pubkey_info['key'], webhook_url) - # Register the secret in the repository url2 = f'https://api.github.com/repos/{self.user_name}/{self.repo_name}/actions/secrets/{secret_name}' payload = { "encrypted_value": encrypted_value, "key_id": key_id } resp2 = requests.put(url2, headers={**headers, 'Content-Type': 'application/json'}, json=payload) - print(f"Secret registered: {secret_name}, {resp2.status_code}, {resp2.text}") + console.print(f"[ INFO ] Secret registered: {secret_name}, {resp2.status_code}, {resp2.text}", style="white") def encrypt(self, public_key: str, secret_value: str) -> str: - """ - Encrypts a secret value using the repository's Base64-encoded public key. - - :param public_key: Repository's public key (Base64-encoded) - :param secret_value: Secret value (plain text) to encrypt - :return: Encrypted secret value (Base64-encoded string) - """ public_key = public.PublicKey(public_key, encoding.Base64Encoder()) sealed_box = public.SealedBox(public_key) encrypted = sealed_box.encrypt(secret_value.encode("utf-8")) diff --git a/src/autofic_core/pr_auto/pr_procedure.py b/src/autofic_core/pr_auto/pr_procedure.py index 6610595..034a52b 100644 --- a/src/autofic_core/pr_auto/pr_procedure.py +++ b/src/autofic_core/pr_auto/pr_procedure.py @@ -25,12 +25,16 @@ import subprocess from pathlib import Path from typing import List +from rich.console import Console from collections import defaultdict + from autofic_core.sast.snippet import BaseSnippet from autofic_core.sast.semgrep.preprocessor import SemgrepPreprocessor from autofic_core.sast.codeql.preprocessor import CodeQLPreprocessor from autofic_core.sast.snykcode.preprocessor import SnykCodePreprocessor +console = Console() + class PRProcedure: """ Handles all modules required for the pull request workflow. @@ -129,7 +133,7 @@ def change_files(self): if os.path.exists(path): subprocess.run(['git', 'reset', '-q', path], check=False) - commit_message = f"[Autofic] {self.vulnerabilities} malicious code detected!!" + commit_message = f"[ AutoFiC ] {self.vulnerabilities} malicious code detected!!" subprocess.run(['git', 'commit', '-m', commit_message], check=True) try: @@ -158,7 +162,7 @@ def generate_pr(self) -> str: Uses vulnerability scan results (by semgrep) to generate a detailed PR body. If llm_generator implemented, then pr_body will add llm_result. """ - print(f"[INFO] Creating PR on {self.user_name}/{self.repo_name}. base branch: {self.base_branch}") + console.print(f"[ INFO ] Creating PR on {self.user_name}/{self.repo_name}. Base branch: {self.base_branch}\n", style="white") pr_url = f"https://api.github.com/repos/{self.user_name}/{self.repo_name}/pulls" if self.tool == "semgrep": snippets = SemgrepPreprocessor.preprocess(self.json_path) @@ -170,7 +174,7 @@ def generate_pr(self) -> str: raise ValueError(f"Unknown tool: {self.tool}") pr_body = self.generate_markdown(snippets) data_post = { - "title": f"[Autofic] Security Patch {datetime.datetime.now().strftime('%Y-%m-%d')}", + "title": f"[ AutoFiC ] Security Patch {datetime.datetime.now().strftime('%Y-%m-%d')}", "head": f"{self.user_name}:{self.branch_name}", "base": self.base_branch, "body": pr_body @@ -254,7 +258,7 @@ def create_pr(self): snippets = SnykCodePreprocessor.preprocess(self.json_path) pr_body = self.generate_markdown(snippets) data_post = { - "title": f"[Autofic] Security Patch {datetime.datetime.now().strftime('%Y-%m-%d')}", + "title": f"[ AutoFiC ] Security Patch {datetime.datetime.now().strftime('%Y-%m-%d')}", "head": f"{self.user_name}:{self.pr_branch}", "base": self.base_branch, "body": pr_body @@ -329,7 +333,7 @@ def generate_markdown_from_llm(llm_path: str) -> str: md = [ "## 🔧 About This Pull Request", - "This patch was automatically created by **[AutoFiC](https://autofic.github.io)**,\nan open-source framework that combines **static analysis tools** with **AI-driven remediation**.", + "This patch was automatically created by **[ AutoFiC ](https://autofic.github.io)**,\nan open-source framework that combines **static analysis tools** with **AI-driven remediation**.", "\nUsing **Semgrep**, **CodeQL**, and **Snyk Code**, AutoFiC detected potential **security flaws** and applied **verified fixes**.", "Each patch includes **contextual explanations** powered by a **large language model** to support **review and decision-making**.", "", diff --git a/src/autofic_core/utils/progress_utils.py b/src/autofic_core/utils/progress_utils.py index 1c6c75f..0797644 100644 --- a/src/autofic_core/utils/progress_utils.py +++ b/src/autofic_core/utils/progress_utils.py @@ -1,10 +1,16 @@ from rich.progress import Progress, SpinnerColumn, BarColumn, TextColumn +from rich.style import Style def create_progress(): return Progress( - SpinnerColumn(), - TextColumn("[bold blue]{task.description}"), - BarColumn(bar_width=None, style="green", complete_style="green"), - TextColumn("[progress.percentage]{task.percentage:>3.0f}%"), + SpinnerColumn(style="cyan"), + TextColumn("{task.description}", style=Style(color="cyan", bold=True), justify="left"), + BarColumn( + bar_width=55, + style=Style(color="blue"), + complete_style=Style(color="bright_blue"), + finished_style=Style(color="bright_blue", bold=True), + ), + TextColumn("{task.percentage:>3.0f}%", style=Style(color="blue", bold=True)), transient=False ) \ No newline at end of file diff --git a/src/autofic_core/utils/ui_utils.py b/src/autofic_core/utils/ui_utils.py index 992bfb0..4ee444c 100644 --- a/src/autofic_core/utils/ui_utils.py +++ b/src/autofic_core/utils/ui_utils.py @@ -5,9 +5,17 @@ console = Console() +def print_divider(title: str): + diamond = "◆" + stage = f"[ {title} ]" + content = f" {diamond} {stage} {diamond} " + + total_width = 90 + line_length = (total_width - len(content)) // 2 + line = "━" * line_length + divider = f"{line}{content}{line}" -def print_divider(title): - console.print(f"\n\n[bold magenta]{'-'*20} [ {title} ] {'-'*20}[/bold magenta]\n\n") + console.print("\n\n" + f"[bold bright_magenta]{divider}[/bold bright_magenta]\n\n") def extract_repo_name(repo_url: str) -> str: @@ -18,11 +26,12 @@ def print_summary(repo_url: str, detected_issues_count: int, output_dir: str, re print_divider("AutoFiC Summary") repo_name = extract_repo_name(repo_url) - console.print(f"✔️ [bold]Target Repository:[/bold] {repo_name}") - console.print(f"✔️ [bold]Files with detected vulnerabilities:[/bold] {detected_issues_count} files") - console.print(f"✔️ [bold]LLM Responses:[/bold] Saved in the 'llm' folder") + console.print(f"📦 [bold]Target Repository:[/bold] {repo_name}") + console.print() + console.print(f"🛡️ [bold]Files with detected vulnerabilities:[/bold] {detected_issues_count} files") + console.print() + console.print(f"🤖 [bold]LLM Responses:[/bold] Saved in the 'llm' folder") - console.print(f"\n[bold magenta]{'━'*64}[/bold magenta]\n") time.sleep(2.0) @@ -66,4 +75,3 @@ def print_help_message(): - The --patch option must be run before using --llm or --llm-retry options. - The --pr option must be run before using --patch option. """) -