#!/usr/bin/env python3 """ stix_validator.py - Validate STIX 2.1 bundles for correctness and completeness. Checks: - Valid JSON structure - Required STIX 2.1 properties on all objects - Valid STIX identifiers (type--UUID format) - Relationship reference integrity (source_ref/target_ref exist) - STIX pattern syntax basics - TLP marking definition correctness - Spec version compliance Usage: python3 stix_validator.py --input report.json python3 stix_validator.py --input report.json --strict python3 stix_validator.py --input report.json --fix --output fixed.json """ from __future__ import annotations import argparse import json import logging import re import sys from pathlib import Path from typing import Optional logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # STIX 2.1 Schema definitions # --------------------------------------------------------------------------- STIX_SDO_TYPES = { "attack-pattern", "campaign", "course-of-action", "grouping", "identity", "indicator", "infrastructure", "intrusion-set", "location", "malware", "malware-analysis", "note", "observed-data", "opinion", "report", "threat-actor", "tool", "vulnerability", } STIX_SRO_TYPES = {"relationship", "sighting"} STIX_META_TYPES = {"marking-definition", "extension-definition", "language-content"} ALL_STIX_TYPES = STIX_SDO_TYPES | STIX_SRO_TYPES | STIX_META_TYPES | {"bundle"} # Required properties per object type REQUIRED_COMMON = {"type", "id", "created", "modified", "spec_version"} REQUIRED_BY_TYPE = { "attack-pattern": {"name"}, "campaign": {"name"}, "identity": {"name"}, "indicator": {"pattern", "pattern_type", "valid_from"}, "malware": {"name", "is_family"}, "report": {"name", "published", "object_refs"}, "threat-actor": {"name"}, "tool": {"name"}, "vulnerability": {"name"}, "relationship": {"relationship_type", "source_ref", "target_ref"}, "sighting": {"sighting_of_ref"}, "marking-definition": {"definition_type", "definition"}, "course-of-action": {"name"}, "grouping": {"context"}, "infrastructure": {"name"}, "intrusion-set": {"name"}, "location": set(), "malware-analysis": {"product", "result"}, "note": {"content", "object_refs"}, "observed-data": {"first_observed", "last_observed", "number_observed"}, "opinion": {"opinion", "object_refs"}, } VALID_RELATIONSHIP_TYPES = { "delivers", "targets", "uses", "attributed-to", "mitigates", "indicates", "variant-of", "impersonates", "derived-from", "duplicate-of", "related-to", "consists-of", "has", "communicates-with", "authored-by", "based-on", "hosts", "located-at", "characterizes", "analysis-of", "av-analysis-of", "static-analysis-of", "dynamic-analysis-of", } VALID_TLP_IDS = { "marking-definition--613f2e26-407d-48c7-9eca-b8e91df99dc9", # TLP:CLEAR "marking-definition--34098fce-860f-48ae-8e50-ebd3cc5e41da", # TLP:GREEN "marking-definition--f88d31f6-486f-44da-b317-01333bde0b82", # TLP:AMBER "marking-definition--826578e1-40a3-4b46-a8d0-ad56a0667d91", # TLP:AMBER+STRICT "marking-definition--5e57c739-391a-4eb3-b6be-7d15ca92d5ed", # TLP:RED } STIX_ID_PATTERN = re.compile( r"^[a-z][a-z0-9-]+--[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$" ) ISO_TIMESTAMP_PATTERN = re.compile( r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?Z$" ) # --------------------------------------------------------------------------- # Validation result # --------------------------------------------------------------------------- class ValidationResult: def __init__(self): self.errors = [] self.warnings = [] self.info = [] def error(self, msg: str, obj_id: str = "") -> None: prefix = f"[{obj_id}] " if obj_id else "" self.errors.append(f"{prefix}{msg}") def warn(self, msg: str, obj_id: str = "") -> None: prefix = f"[{obj_id}] " if obj_id else "" self.warnings.append(f"{prefix}{msg}") def add_info(self, msg: str) -> None: self.info.append(msg) @property def is_valid(self) -> bool: return len(self.errors) == 0 def summary(self) -> str: lines = ["=" * 60, "STIX Validation Report", "=" * 60, ""] if self.is_valid: lines.append("RESULT: VALID") else: lines.append("RESULT: INVALID") lines.append(f" Errors: {len(self.errors)}") lines.append(f" Warnings: {len(self.warnings)}") lines.append("") if self.info: lines.append("--- Bundle Info ---") for i in self.info: lines.append(f" {i}") lines.append("") if self.errors: lines.append("--- Errors ---") for e in self.errors: lines.append(f" ERROR: {e}") lines.append("") if self.warnings: lines.append("--- Warnings ---") for w in self.warnings: lines.append(f" WARN: {w}") lines.append("") return "\n".join(lines) # --------------------------------------------------------------------------- # Validators # --------------------------------------------------------------------------- def validate_stix_id(stix_id: str, expected_type: str = "") -> Optional[str]: """Validate a STIX identifier. Returns error message or None.""" if not STIX_ID_PATTERN.match(stix_id): return f"Invalid STIX ID format: {stix_id}" if expected_type: prefix = stix_id.split("--")[0] if prefix != expected_type: return f"ID prefix '{prefix}' does not match type '{expected_type}'" return None def validate_timestamp(ts: str) -> Optional[str]: """Validate an ISO 8601 timestamp. Returns error message or None.""" if not ISO_TIMESTAMP_PATTERN.match(ts): return f"Invalid timestamp format: {ts} (expected ISO 8601 with Z suffix)" return None def validate_pattern_syntax(pattern: str) -> Optional[str]: """Basic STIX pattern syntax validation. Returns error message or None.""" if not pattern: return "Pattern is empty" # Check balanced brackets if pattern.count("[") != pattern.count("]"): return f"Unbalanced brackets in pattern: {pattern}" if pattern.count("'") % 2 != 0: return f"Unbalanced single quotes in pattern: {pattern}" # Check for at least one comparison expression if "[" not in pattern or "]" not in pattern: return f"Pattern missing observation expression brackets: {pattern}" # Check for valid comparison operators valid_ops = ["=", "!=", ">", "<", ">=", "<=", "IN", "LIKE", "MATCHES", "ISSUBSET", "ISSUPERSET"] has_operator = any(f" {op} " in pattern or f" {op} " in pattern.upper() for op in valid_ops) if not has_operator: return f"Pattern missing comparison operator: {pattern}" return None def validate_bundle(data: dict, strict: bool = False) -> ValidationResult: """Validate a complete STIX 2.1 bundle.""" result = ValidationResult() # Check bundle type if data.get("type") != "bundle": result.error("Top-level 'type' must be 'bundle'") return result # Check bundle ID bundle_id = data.get("id", "") if not bundle_id.startswith("bundle--"): result.error(f"Bundle ID must start with 'bundle--': {bundle_id}") # Check objects array objects = data.get("objects") if not objects: result.error("Bundle has no 'objects' array or it is empty") return result if not isinstance(objects, list): result.error("'objects' must be an array") return result result.add_info(f"Total objects: {len(objects)}") # Collect all IDs for reference checking all_ids = set() obj_types = {} for obj in objects: obj_id = obj.get("id", "") all_ids.add(obj_id) obj_type = obj.get("type", "unknown") obj_types[obj_type] = obj_types.get(obj_type, 0) + 1 result.add_info(f"Object types: {obj_types}") # Validate each object for obj in objects: obj_type = obj.get("type", "") obj_id = obj.get("id", "") # Validate type if not obj_type: result.error("Object missing 'type' property", obj_id) continue if obj_type not in ALL_STIX_TYPES and not obj_type.startswith("x-"): result.warn(f"Unknown STIX object type: {obj_type}", obj_id) # Skip deeper validation for marking definitions (simpler schema) if obj_type == "marking-definition": _validate_marking(obj, result) continue if obj_type == "bundle": continue # Validate ID if obj_id: id_error = validate_stix_id(obj_id, obj_type) if id_error: result.error(id_error, obj_id) else: result.error(f"Object of type '{obj_type}' missing 'id'") continue # Validate spec_version spec = obj.get("spec_version", "") if spec and spec != "2.1": result.warn(f"spec_version is '{spec}', expected '2.1'", obj_id) elif not spec and obj_type not in STIX_META_TYPES: if strict: result.error("Missing 'spec_version'", obj_id) else: result.warn("Missing 'spec_version'", obj_id) # Validate timestamps for ts_field in ("created", "modified"): ts = obj.get(ts_field, "") if ts: ts_error = validate_timestamp(ts) if ts_error: result.error(f"{ts_field}: {ts_error}", obj_id) elif obj_type not in STIX_META_TYPES: if strict: result.error(f"Missing '{ts_field}'", obj_id) # Validate required properties by type required = REQUIRED_BY_TYPE.get(obj_type, set()) for prop in required: if prop not in obj: result.error(f"Missing required property '{prop}'", obj_id) # Validate relationships if obj_type == "relationship": _validate_relationship(obj, all_ids, result) # Validate indicators if obj_type == "indicator": _validate_indicator(obj, result) # Validate object_marking_refs marking_refs = obj.get("object_marking_refs", []) for ref in marking_refs: if ref not in all_ids: result.warn(f"Marking reference '{ref}' not found in bundle", obj_id) # Validate object_refs (for reports, notes, etc.) obj_refs = obj.get("object_refs", []) for ref in obj_refs: if ref not in all_ids: result.warn(f"Object reference '{ref}' not found in bundle", obj_id) return result def _validate_relationship(obj: dict, all_ids: set, result: ValidationResult): """Validate a Relationship SRO.""" obj_id = obj.get("id", "") rel_type = obj.get("relationship_type", "") source_ref = obj.get("source_ref", "") target_ref = obj.get("target_ref", "") if rel_type and rel_type not in VALID_RELATIONSHIP_TYPES: result.warn(f"Non-standard relationship type: '{rel_type}'", obj_id) if source_ref and source_ref not in all_ids: result.error(f"source_ref '{source_ref}' not found in bundle", obj_id) if target_ref and target_ref not in all_ids: result.error(f"target_ref '{target_ref}' not found in bundle", obj_id) if source_ref == target_ref and source_ref: result.warn("source_ref and target_ref are the same object", obj_id) def _validate_indicator(obj: dict, result: ValidationResult): """Validate an Indicator SDO.""" obj_id = obj.get("id", "") pattern = obj.get("pattern", "") pattern_type = obj.get("pattern_type", "") if pattern and pattern_type == "stix": pattern_error = validate_pattern_syntax(pattern) if pattern_error: result.error(f"Pattern syntax: {pattern_error}", obj_id) valid_from = obj.get("valid_from", "") valid_until = obj.get("valid_until", "") if valid_from and valid_until: if valid_until <= valid_from: result.warn("valid_until is not after valid_from", obj_id) def _validate_marking(obj: dict, result: ValidationResult): """Validate a marking-definition object.""" obj_id = obj.get("id", "") def_type = obj.get("definition_type", "") if def_type == "tlp": if obj_id not in VALID_TLP_IDS: result.warn(f"Non-standard TLP marking definition ID: {obj_id}", obj_id) definition = obj.get("definition", {}) tlp_value = definition.get("tlp", "") if tlp_value not in ("clear", "white", "green", "amber", "amber+strict", "red"): result.error(f"Invalid TLP value: '{tlp_value}'", obj_id) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Validate STIX 2.1 bundles for correctness and completeness.", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: %(prog)s --input report.json %(prog)s --input report.json --strict %(prog)s --input report.json --json """, ) parser.add_argument("--input", "-i", required=True, help="STIX bundle JSON file to validate") parser.add_argument("--strict", action="store_true", help="Strict validation (warnings become errors)") parser.add_argument("--json", action="store_true", help="Output results as JSON") parser.add_argument("--verbose", "-v", action="store_true", help="Verbose output") parser.add_argument("--format", default="json", choices=["json", "text", "csv"], help="Output format (default: json)") return parser.parse_args() def main() -> None: args = parse_arguments() logging.basicConfig( level=logging.DEBUG if args.verbose else logging.INFO, format="%(levelname)s: %(message)s", ) input_path = Path(args.input) if not input_path.is_file(): logger.error(f"File not found: {input_path}") sys.exit(1) # Load JSON try: data = json.loads(input_path.read_text()) except json.JSONDecodeError as e: logger.error(f"Invalid JSON: {e}") sys.exit(1) # Validate result = validate_bundle(data, strict=args.strict) # Output if args.json: output = { "valid": result.is_valid, "errors": result.errors, "warnings": result.warnings, "info": result.info, } print(json.dumps(output, indent=2)) else: print(result.summary()) sys.exit(0 if result.is_valid else 1) if __name__ == "__main__": main()