feat(connector-linter): add VC3xx code checks

Author: jabesq

shared/connector_linter/connector_linter/checks/_helpers.py

@@ -0,0 +1,48 @@
+import ast
+from pathlib import Path
+
+from connector_linter.models import ConnectorContext
+
+# ---------------------------------------------------------------------------
+# Source reading
+# ---------------------------------------------------------------------------
+
+
+def read_all_python_sources(ctx: ConnectorContext) -> dict[Path, str]:
+    """Read all Python source files from the connector's src/ directory."""
+    sources: dict[Path, str] = {}
+    # Convention: all connector Python code lives under <connector>/src/
+    src_dir = ctx.path / "src"
+    if not src_dir.exists():
+        return sources
+    for py_file in src_dir.rglob("*.py"):
+        # Key by relative path (from connector root) for portable reporting
+        rel_path = py_file.relative_to(ctx.path)
+        try:
+            # errors="replace" avoids UnicodeDecodeError on malformed files
+            sources[rel_path] = py_file.read_text(encoding="utf-8", errors="replace")
+        except OSError:
+            # Skip unreadable files (permissions, broken symlinks, etc.)
+            continue
+    return sources
+
+
+# ---------------------------------------------------------------------------
+# AST helpers — structural analysis of Python source
+# ---------------------------------------------------------------------------
+
+
+def parse_sources(sources: dict[Path, str]) -> dict[Path, ast.Module]:
+    """Parse all source files into AST modules.
+
+    Files that fail to parse (syntax errors) are silently skipped.
+    """
+    trees: dict[Path, ast.Module] = {}
+    for file_path, content in sources.items():
+        try:
+            trees[file_path] = ast.parse(content, filename=str(file_path))
+        except SyntaxError:
+            # Silently skip files with syntax errors — they can't be analyzed
+            # structurally, but other checks (regex-based) may still find issues.
+            continue
+    return trees

shared/connector_linter/connector_linter/checks/vc3xx_code/__init__.py

@@ -1 +1,27 @@
-"""VC3xx — Code quality checks."""
+"""VC3xx — Code quality checks.
+
+VC301: Connector must define an author identity.
+VC302: Author must be referenced on STIX entities (created_by_ref).
+VC303: CONNECTOR_TYPE must be defined in application code, not read from env.
+VC304: Ensure TLP markings are checked (check_max_tlp).
+VC305: Connector must implement Base Settings from connectors-sdk.
+VC306: Connector log level should default to 'error'.
+VC307: Except blocks should use error/warning logging, not debug/info.
+VC308: Main entry point must use traceback for error handling.
+VC309: Connector must use only absolute imports, no relative imports.
+VC310: External references must not be added by default to non-Identity objects.
+VC311: Connector should use TLP markings on entities with appropriate level.
+VC312: send_stix2_bundle must use cleanup_inconsistent_bundle=True.
+VC313: STIX SDO/SRO objects must use pycti.XXX.generate_id() for deterministic IDs.
+VC314: External-import connectors must use schedule_process or schedule_iso.
+VC315: Connector must call initiate_work before processing.
+VC316: Connector must close work with to_processed after processing.
+VC317: initiate_work should only be called when data is available.
+VC318: Internal-enrichment connectors must use helper.listen().
+VC319: Enrichment connector must return original bundle when not in scope.
+VC320: Enrichment connector must enforce TLP access control.
+VC321: Enrichment connector must be playbook-compatible.
+VC322: Enrichment connector must read data['stix_objects'] (former bundle).
+VC323: Stream connectors must use helper.listen_stream().
+VC324: Relationship should not set both start_time and stop_time.
+"""

shared/connector_linter/connector_linter/checks/vc3xx_code/vc301_author_defined.py

@@ -0,0 +1,166 @@
+"""VC301 — Connector must define an author identity.
+
+Uses AST to detect author identity definitions by looking for constructor
+calls (``Identity(...)``, ``OrganizationAuthor(...)``, ``stix2.Identity(...)``)
+and API calls (``helper.api.identity.create(...)``).
+
+Import validation for bare ``Identity(...)`` calls is also AST-based:
+the call is only counted when ``Identity`` is imported from ``stix2``
+or ``pycti`` (not some unrelated class).
+"""
+
+import ast
+from pathlib import Path
+
+from connector_linter.models import (
+    CheckFinding,
+    ConnectorContext,
+    ConnectorType,
+    Severity,
+    no_python_sources_finding,
+)
+from connector_linter.registry import CheckRegistry
+
+# ---------------------------------------------------------------------------
+# Author-definition patterns (AST-based)
+# ---------------------------------------------------------------------------
+
+# Function/constructor names that unambiguously define an author
+_UNAMBIGUOUS_AUTHOR_CALLS = {"OrganizationAuthor"}
+
+# Import modules that confirm bare Identity() comes from stix2/pycti
+_IDENTITY_MODULES = {"stix2", "pycti"}
+
+
+def _has_identity_import(trees: dict[Path, ast.Module]) -> bool:
+    """Check if any source file imports Identity from stix2 or pycti."""
+    for tree in trees.values():
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                # from stix2[.xxx] import Identity / from pycti[.xxx] import Identity
+                mod_root = node.module.split(".")[0]
+                if mod_root in _IDENTITY_MODULES:
+                    for alias in node.names:
+                        if alias.name == "Identity":
+                            return True
+            elif isinstance(node, ast.Import):
+                # import stix2  (Identity accessed as stix2.Identity)
+                for alias in node.names:
+                    if alias.name.split(".")[0] in _IDENTITY_MODULES:
+                        return True
+    return False
+
+
+def _is_author_call(node: ast.Call, identity_imported: bool) -> bool:
+    """Determine if an ast.Call node represents an author definition.
+
+    Recognized patterns:
+    1. OrganizationAuthor(...)          — connectors-sdk
+    2. stix2.Identity(...)              — qualified stix2 constructor
+    3. Identity(...)                    — bare, only if imported from stix2/pycti
+    4. *.api.identity.create(...)       — legacy pycti API call
+    """
+    func = node.func
+
+    # Pattern 1 & 3: bare function call — OrganizationAuthor(...) or Identity(...)
+    if isinstance(func, ast.Name):
+        if func.id in _UNAMBIGUOUS_AUTHOR_CALLS:
+            return True
+        if func.id == "Identity" and identity_imported:
+            return True
+
+    # Patterns 2 & 4: attribute-based calls
+    if isinstance(func, ast.Attribute):
+        # Pattern 2: stix2.Identity(...)
+        if func.attr == "Identity" and isinstance(func.value, ast.Name):
+            if func.value.id == "stix2":
+                return True
+
+        # Pattern 4: *.api.identity.create(...)
+        if (
+            func.attr == "create"
+            and isinstance(func.value, ast.Attribute)
+            and func.value.attr == "identity"
+            and isinstance(func.value.value, ast.Attribute)
+            and func.value.value.attr == "api"
+        ):
+            return True
+
+    return False
+
+
+def find_author_definitions(
+    sources: dict[Path, str],
+    trees: dict[Path, ast.Module],
+) -> list[tuple[Path, int, str]]:
+    """Find author definition locations using AST analysis.
+
+    Args:
+        sources: Raw Python source content keyed by relative path (used for
+            line-text extraction in findings).
+        trees:   Pre-parsed AST modules (e.g. ``ctx.python_trees``).  Passing
+            the cached property avoids redundant parsing across checks.
+
+    Returns:
+        List of (file_path, line_number, matched_line_text).
+    """
+    identity_imported = _has_identity_import(trees)
+
+    hits: list[tuple[Path, int, str]] = []
+    for file_path, tree in trees.items():
+        content_lines = sources[file_path].splitlines()
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call) and _is_author_call(node, identity_imported):
+                line_text = (
+                    content_lines[node.lineno - 1].strip()
+                    if node.lineno <= len(content_lines)
+                    else ""
+                )
+                hits.append((file_path, node.lineno, line_text))
+
+    return hits
+
+
+@CheckRegistry.register(
+    code="VC301",
+    name="author-defined",
+    description="Connector must define an author identity",
+    severity=Severity.ERROR,
+    applicable_types={
+        ConnectorType.INTERNAL_ENRICHMENT,
+        ConnectorType.EXTERNAL_IMPORT,
+        ConnectorType.INTERNAL_IMPORT_FILE,
+    },
+)
+def check_author_defined(ctx: ConnectorContext) -> list[CheckFinding]:
+    """Check that the connector defines an author identity somewhere in its source."""
+    sources = ctx.python_sources
+
+    if not sources:
+        return [no_python_sources_finding()]
+
+    hits = find_author_definitions(sources, ctx.python_trees)
+
+    if hits:
+        file_path, line, _ = hits[0]
+        return [
+            CheckFinding(
+                message="Author identity defined",
+                severity=Severity.INFO,
+                file_path=file_path,
+                line=line,
+            ),
+        ]
+
+    return [
+        CheckFinding(
+            message="No author identity definition found in connector source",
+            severity=Severity.ERROR,
+            suggestion=(
+                "Define an author using one of: "
+                "stix2.Identity(name=..., identity_class='organization'), "
+                "OrganizationAuthor(name=...), "
+                "or self.helper.api.identity.create(type='Organization', name=...)"
+            ),
+        ),
+    ]

shared/connector_linter/connector_linter/checks/vc3xx_code/vc302_author_referenced.py

@@ -0,0 +1,117 @@
+"""VC302 — Author must be referenced on STIX entities (created_by_ref).
+
+Uses AST to detect ``author=`` as a keyword argument in function calls
+(avoids false positives from variable assignments like ``author = "John"``).
+Also detects ``created_by_ref=`` and ``x_opencti_created_by_ref`` via AST.
+"""
+
+import ast
+from pathlib import Path
+
+from connector_linter.checks.vc3xx_code.vc301_author_defined import (
+    find_author_definitions,
+)
+from connector_linter.models import (
+    CheckFinding,
+    ConnectorContext,
+    ConnectorType,
+    Severity,
+    no_python_sources_finding,
+)
+from connector_linter.registry import CheckRegistry
+
+# ---------------------------------------------------------------------------
+# Keyword argument names that attach an author identity to STIX entities:
+#
+#   created_by_ref           — standard STIX 2.1 field (SDOs)
+#   author                   — connectors-sdk model parameter
+#   x_opencti_created_by_ref — OpenCTI custom property (used on observables/SCOs)
+# ---------------------------------------------------------------------------
+_AUTHOR_KWARGS = {"created_by_ref", "author", "x_opencti_created_by_ref"}
+
+
+def _find_author_references(
+    trees: dict[Path, ast.Module],
+) -> list[tuple[Path, int, str]]:
+    """Find keyword arguments that reference an author on STIX entities.
+
+    Detects ``created_by_ref=``, ``author=``, and ``x_opencti_created_by_ref``
+    as keyword arguments in function/constructor calls — not plain assignments.
+    """
+    hits: list[tuple[Path, int, str]] = []
+    for file_path, tree in trees.items():
+        for node in ast.walk(tree):
+            # Only check Call nodes — we want keyword arguments in function/
+            # constructor calls, not standalone assignments like `author = "John"`
+            if not isinstance(node, ast.Call):
+                continue
+            for kw in node.keywords:
+                if kw.arg in _AUTHOR_KWARGS:
+                    # Use the keyword node's own lineno when available;
+                    # fall back to the call node's lineno otherwise
+                    hits.append(
+                        (
+                            file_path,
+                            getattr(kw, "lineno", node.lineno) or node.lineno,
+                            kw.arg,
+                        ),
+                    )
+    return hits
+
+
+@CheckRegistry.register(
+    code="VC302",
+    name="author-referenced-on-entities",
+    description="Author must be referenced on STIX entities (created_by_ref)",
+    severity=Severity.ERROR,
+    applicable_types={
+        ConnectorType.INTERNAL_ENRICHMENT,
+        ConnectorType.EXTERNAL_IMPORT,
+        ConnectorType.INTERNAL_IMPORT_FILE,
+    },
+)
+def check_author_referenced(ctx: ConnectorContext) -> list[CheckFinding]:
+    """Check that created_by_ref or author= is used to attach author to entities."""
+    sources = ctx.python_sources
+
+    if not sources:
+        return [no_python_sources_finding()]
+
+    # Dependency: an author must be defined first (VC301) before it can be referenced.
+    # If no author definition exists, we fail with a clear message pointing to VC301.
+    author_hits = find_author_definitions(sources, ctx.python_trees)
+    if not author_hits:
+        return [
+            CheckFinding(
+                message="No author defined — cannot reference author on entities",
+                severity=Severity.ERROR,
+                suggestion="Define an author first (see VC301), then use created_by_ref= on STIX objects",
+            ),
+        ]
+
+    # Author exists — now check if it's actually referenced on STIX entities (AST-based)
+    trees = ctx.python_trees
+    ref_hits = _find_author_references(trees)
+
+    if ref_hits:
+        file_path, line, _kwarg = ref_hits[0]
+        return [
+            CheckFinding(
+                message=f"Author referenced on entities ({len(ref_hits)} occurrence(s) found)",
+                severity=Severity.INFO,
+                file_path=file_path,
+                line=line,
+            ),
+        ]
+
+    return [
+        CheckFinding(
+            message="Author is defined but never referenced on STIX entities",
+            severity=Severity.ERROR,
+            suggestion=(
+                "Use created_by_ref=self.author.id on SDOs, "
+                "or x_opencti_created_by_ref in custom_properties for observables, "
+                "or author= parameter when using connectors-sdk models"
+            ),
+        ),
+    ]