Initial plan for fixing linter errors

Co-authored-by: kevinbackhouse <[email protected]>

Author: Copilot

src/seclab_taskflows/mcp_servers/alert_results_models.py

@@ -1,10 +1,12 @@
 # SPDX-FileCopyrightText: 2025 GitHub
 # SPDX-License-Identifier: MIT
 
-from sqlalchemy import String, Text, Integer, ForeignKey, Column
-from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped, relationship
 from typing import Optional
 
+from sqlalchemy import Column, ForeignKey, Integer, Text
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+
+
 class Base(DeclarativeBase):
     pass
 

src/seclab_taskflows/mcp_servers/codeql_python/codeql_sqlite_models.py

@@ -1,10 +1,12 @@
 # SPDX-FileCopyrightText: 2025 GitHub
 # SPDX-License-Identifier: MIT
 
-from sqlalchemy import Text
-from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
 from typing import Optional
 
+from sqlalchemy import Text
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+
+
 class Base(DeclarativeBase):
     pass
 

src/seclab_taskflows/mcp_servers/codeql_python/mcp_server.py

@@ -3,29 +3,31 @@
 
 
 import logging
+
 logging.basicConfig(
     level=logging.DEBUG,
     format='%(asctime)s - %(levelname)s - %(message)s',
     filename='logs/mcp_codeql_python.log',
     filemode='a'
 )
-from seclab_taskflow_agent.mcp_servers.codeql.client import run_query, _debug_log
-# from seclab_taskflow_agent.path_utils import mcp_data_dir
-
-from pydantic import Field
-#from mcp.server.fastmcp import FastMCP, Context
-from fastmcp import FastMCP # use FastMCP 2.0
-from pathlib import Path
-import os
 import csv
+import importlib.resources
 import json
+import os
+import subprocess
+from pathlib import Path
+
+#from mcp.server.fastmcp import FastMCP, Context
+from fastmcp import FastMCP  # use FastMCP 2.0
+
+# from seclab_taskflow_agent.path_utils import mcp_data_dir
+from pydantic import Field
+from seclab_taskflow_agent.mcp_servers.codeql.client import _debug_log, run_query
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
-import subprocess
-import importlib.resources
 
-from .codeql_sqlite_models import Base, Source
 from ..utils import process_repo
+from .codeql_sqlite_models import Base, Source
 
 MEMORY = Path(os.getenv('DATA_DIR', default='/app/data'))
 CODEQL_DBS_BASE_PATH = Path(os.getenv('CODEQL_DBS_BASE_PATH', default='/app/data'))
@@ -96,13 +98,12 @@ def store_new_source(self, repo, source_location, line, source_type, notes, upda
                 existing.notes = (existing.notes or "") + notes
                 session.commit()
                 return f"Updated notes for source at {source_location}, line {line} in {repo}."
-            else:
-                if update:
-                    return f"No source exists at repo {repo}, location {source_location}, line {line} to update."
-                new_source = Source(repo = repo,  source_location = source_location, line = line, source_type = source_type, notes = notes)
-                session.add(new_source)
-                session.commit()
-                return f"Added new source for {source_location} in {repo}."
+            if update:
+                return f"No source exists at repo {repo}, location {source_location}, line {line} to update."
+            new_source = Source(repo = repo,  source_location = source_location, line = line, source_type = source_type, notes = notes)
+            session.add(new_source)
+            session.commit()
+            return f"Added new source for {source_location} in {repo}."
 
     def get_sources(self, repo):
         with Session(self.engine) as session:
@@ -221,5 +222,5 @@ def clear_codeql_repo(owner: str = Field(description="The owner of the GitHub re
     if not os.path.isdir('/.codeql/packages/codeql/python-all'):
         pack_path = importlib.resources.files('seclab_taskflows.mcp_servers.codeql_python.queries').joinpath('mcp-python')
         print(f"Installing CodeQL pack from {pack_path}")
-        subprocess.run(["codeql", "pack", "install", pack_path])
+        subprocess.run(["codeql", "pack", "install", pack_path], check=False)
     mcp.run(show_banner=False, transport="http", host="127.0.0.1", port=9998)

src/seclab_taskflows/mcp_servers/gh_actions.py

@@ -2,23 +2,24 @@
 # SPDX-License-Identifier: MIT
 
 import logging
+
 logging.basicConfig(
     level=logging.DEBUG,
     format='%(asctime)s - %(levelname)s - %(message)s',
     filename='logs/mcp_gh_actions.log',
     filemode='a'
 )
 
-from fastmcp import FastMCP
-from pydantic import Field
-import httpx
 import json
 import os
+from pathlib import Path
+
+import httpx
 import yaml
-from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
+from fastmcp import FastMCP
+from pydantic import Field
 from sqlalchemy import create_engine
-from sqlalchemy.orm import Session
-from pathlib import Path
+from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 
 
 class Base(DeclarativeBase):
@@ -218,9 +219,9 @@ async def check_workflow_reusable(
     for trigger in triggers:
         if isinstance(trigger, str) and trigger == "workflow_call":
             return "This workflow is reusable as a workflow call."
-        elif isinstance(trigger, dict):
+        if isinstance(trigger, dict):
             for k, v in trigger.items():
-                if 'workflow_call' == k:
+                if k == 'workflow_call':
                     return "This workflow is reusable."
     return "This workflow is not reusable."
 
@@ -236,19 +237,15 @@ async def get_high_privileged_workflow_triggers(
     results = []
     for trigger in triggers:
         if isinstance(trigger, str):
-            if trigger in high_privileged_triggers:
-                results.append(trigger)
-            elif trigger == 'workflow_run':
+            if trigger in high_privileged_triggers or trigger == 'workflow_run':
                 results.append(trigger)
         elif isinstance(trigger, dict):
             this_results = {}
             for k, v in trigger.items():
                 if k in high_privileged_triggers:
                     this_results[k] = v
                 elif k == 'workflow_run':
-                    if not v or isinstance(v, str):
-                        this_results[k] = v
-                    elif isinstance(v, dict) and not 'branches' in v:
+                    if not v or isinstance(v, str) or (isinstance(v, dict) and 'branches' not in v):
                         this_results[k] = v
             if this_results:
                 results.append(this_results)
@@ -293,7 +290,7 @@ async def get_workflow_user(
             if action_name in use:
                 actual_name[use] = []
         for i, line in enumerate(lines):
-            for use in actual_name.keys():
+            for use in actual_name:
                 if use in line:
                     actual_name[use].append(i + 1)
         for use, line_numbers in actual_name.items():
@@ -316,7 +313,7 @@ async def get_workflow_user(
                 workflow_use = WorkflowUses(**result)
                 session.add(workflow_use)
             session.commit()
-        return f"Search results saved to database."
+        return "Search results saved to database."
     return json.dumps(results)
 
 @mcp.tool()

src/seclab_taskflows/mcp_servers/gh_code_scanning.py

@@ -2,26 +2,28 @@
 # SPDX-License-Identifier: MIT
 
 import logging
+
 logging.basicConfig(
     level=logging.DEBUG,
     format='%(asctime)s - %(levelname)s - %(message)s',
     filename='logs/mcp_gh_code_scanning.log',
     filemode='a'
 )
-from fastmcp import FastMCP
-from pydantic import Field
-import httpx
-import aiofiles
 import json
 import os
 import re
-from urllib.parse import urlparse, parse_qs
-from pathlib import Path
 import zipfile
+from pathlib import Path
+from urllib.parse import parse_qs, urlparse
+
+import aiofiles
+import httpx
+from fastmcp import FastMCP
+from pydantic import Field
 from sqlalchemy import create_engine
 from sqlalchemy.orm import Session
 
-from .alert_results_models import AlertResults, AlertFlowGraph, Base
+from .alert_results_models import AlertFlowGraph, AlertResults, Base
 
 mcp = FastMCP("GitHubCodeScanning")
 
@@ -158,7 +160,7 @@ async def fetch_alerts_to_sql(
     ) -> str:
     """Fetch all code scanning alerts for a specific repository and store them in a SQL database."""
     results = await fetch_alerts_from_gh(owner, repo, state, rule)
-    sql_db_path = f"sqlite:///{ALERT_RESULTS_DIR}/alert_results.db" 
+    sql_db_path = f"sqlite:///{ALERT_RESULTS_DIR}/alert_results.db"
     if isinstance(results, str) or not results:
         return results
     engine = create_engine(sql_db_path, echo=False)

src/seclab_taskflows/mcp_servers/gh_file_viewer.py

@@ -2,26 +2,27 @@
 # SPDX-License-Identifier: MIT
 
 import logging
+
 logging.basicConfig(
     level=logging.DEBUG,
     format='%(asctime)s - %(levelname)s - %(message)s',
     filename='logs/mcp_gh_file_viewer.log',
     filemode='a'
 )
 
-from fastmcp import FastMCP
-from pydantic import Field
-import httpx
 import json
 import os
-from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped
-from sqlalchemy import create_engine
-from sqlalchemy.orm import Session
-from typing import Optional
+import tempfile
+import zipfile
 from pathlib import Path
+
 import aiofiles
-import zipfile
-import tempfile
+import httpx
+from fastmcp import FastMCP
+from pydantic import Field
+from sqlalchemy import create_engine
+from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
+
 
 class Base(DeclarativeBase):
     pass
@@ -112,7 +113,7 @@ def search_zipfile(database_path, term):
                 for i, line in enumerate(f):
                     if term in str(line):
                         filename = remove_root_dir(entry.filename)
-                        if not filename in results:
+                        if filename not in results:
                             results[filename] = [i+1]
                         else:
                             results[filename].append(i+1)
@@ -154,8 +155,7 @@ async def get_file_lines_from_gh(
     if isinstance(r, str):
         return r
     lines = r.text.splitlines()
-    if start_line < 1:
-        start_line = 1
+    start_line = max(start_line, 1)
     if length < 1:
         length = 10
     lines = lines[start_line-1:start_line-1+length]
@@ -217,7 +217,7 @@ async def search_files_from_gh(
                 search_result = SearchResults(**result)
                 session.add(search_result)
             session.commit()
-        return f"Search results saved to database."
+        return "Search results saved to database."
     return json.dumps(results)
 
 @mcp.tool()

src/seclab_taskflows/mcp_servers/ghsa.py

@@ -1,16 +1,19 @@
 import logging
+
 logging.basicConfig(
     level=logging.DEBUG,
     format='%(asctime)s - %(levelname)s - %(message)s',
     filename='logs/mcp_ghsa.log',
     filemode='a'
 )
 
+import json
+import re
+from urllib.parse import parse_qs, urlparse
+
 from fastmcp import FastMCP
 from pydantic import Field
-import re
-import json
-from urllib.parse import urlparse, parse_qs
+
 from .gh_code_scanning import call_api
 
 mcp = FastMCP("GitHubRepoAdvisories")

src/seclab_taskflows/mcp_servers/local_file_viewer.py

@@ -2,23 +2,21 @@
 # SPDX-License-Identifier: MIT
 
 import logging
+
 logging.basicConfig(
     level=logging.DEBUG,
     format='%(asctime)s - %(levelname)s - %(message)s',
     filename='logs/mcp_local_file_viewer.log',
     filemode='a'
 )
 
-from fastmcp import FastMCP
-from pydantic import Field
-import httpx
 import json
 import os
-from pathlib import Path
-import aiofiles
 import zipfile
-import tempfile
+from pathlib import Path
 
+from fastmcp import FastMCP
+from pydantic import Field
 
 mcp = FastMCP("LocalFileViewer")
 
@@ -61,7 +59,7 @@ def search_zipfile(database_path, term, search_dir = None):
                 for i, line in enumerate(f):
                     if term in str(line):
                         filename = remove_root_dir(entry.filename)
-                        if not filename in results:
+                        if filename not in results:
                             results[filename] = [i+1]
                         else:
                             results[filename].append(i+1)
@@ -126,8 +124,7 @@ async def get_file_lines(
     if not source_path or not source_path.exists():
         return f"Invalid {owner} and {repo}. Check that the input is correct or try to fetch the repo from gh first."
     lines = get_file(source_path, path)
-    if start_line < 1:
-        start_line = 1
+    start_line = max(start_line, 1)
     if length < 1:
         length = 10
     lines = lines[start_line-1:start_line-1+length]

src/seclab_taskflows/mcp_servers/local_gh_resources.py

@@ -2,23 +2,21 @@
 # SPDX-License-Identifier: MIT
 
 import logging
+
 logging.basicConfig(
     level=logging.DEBUG,
     format='%(asctime)s - %(levelname)s - %(message)s',
     filename='logs/mcp_local_gh_resources.log',
     filemode='a'
 )
 
-from fastmcp import FastMCP
-from pydantic import Field
-import httpx
 import json
 import os
 from pathlib import Path
-import aiofiles
-import zipfile
-import tempfile
 
+import aiofiles
+import httpx
+from fastmcp import FastMCP
 
 mcp = FastMCP("LocalGHResources")