diff --git a/README.md b/README.md index b22fe5ec..11ce3aa8 100644 --- a/README.md +++ b/README.md @@ -74,6 +74,15 @@ If you provide none of the flags in the `ANALYSIS SELECTION` section of the `--h The `--streams` output is the one used to hunt for patterns in the embedded bytes and can be _extremely_ verbose depending on the `--quote-char` options chosen (or not chosen) and contents of the PDF. [The Yaralyzer](https://github.com/michelcrypt4d4mus/yaralyzer) handles this task; if you want to hunt for patterns in the bytes other than bytes surrounded by backticks/frontslashes/brackets/quotes/etc. you may want to use The Yaralyzer directly. As The Yaralyzer is a prequisite for The Pdfalyzer you may already have the `yaralyze` command installed and available. +### Exporting to JSON +Use the `-json` flag to export analysis results to structured JSON files: + +```sh +pdfalyze suspicious.pdf -json --output-dir ./analysis +``` + +This will create JSON files for each analysis type (document info, tree structure, fonts, etc.) preserving the complete PDF structure for programmatic processing. + ### Setting Command Line Options Permanently With A `.pdfalyzer` File When you run `pdfalyze` on some PDF the tool will check for a file called `.pdfalyzer` first in the current directory and then in the home directory. If it finds a file in either such place it will load configuration options from it. Documentation on the options that can be configured with these files lives in [`.pdfalyzer.example`](.pdfalyzer.example) which doubles as an example file you can copy into place and edit to your needs. Handy if you find yourself typing the same command line options over and over again. @@ -141,7 +150,7 @@ for backtick_quoted_string in font.binary_scanner.extract_backtick_quoted_bytes( ------------- # Example Output -The Pdfalyzer can export visualizations to HTML, ANSI colored text, and SVG images using the file export functionality that comes with [Rich](https://github.com/Textualize/rich). SVGs can be turned into `png` format images with a tool like Inkscape or `cairosvg` (Inkscape works a lot better in our experience). See `pdfalyze --help` for the specifics. +The Pdfalyzer can export visualizations to HTML, ANSI colored text, SVG images, and JSON data files using the file export functionality that comes with [Rich](https://github.com/Textualize/rich). SVGs can be turned into `png` format images with a tool like Inkscape or `cairosvg` (Inkscape works a lot better in our experience). JSON export preserves the complete PDF structure for programmatic analysis. See `pdfalyze --help` for the specifics. ## Basic Tree View diff --git a/pdfalyzer/__init__.py b/pdfalyzer/__init__.py index 9c7e9029..0e15f80f 100644 --- a/pdfalyzer/__init__.py +++ b/pdfalyzer/__init__.py @@ -1,5 +1,6 @@ import code import sys +from datetime import datetime from os import environ, getcwd, path from pathlib import Path @@ -27,6 +28,7 @@ from pdfalyzer.helpers.filesystem_helper import file_size_in_mb, set_max_open_files from pdfalyzer.helpers.rich_text_helper import print_highlighted +from pdfalyzer.output.json_exporter import JsonExporter from pdfalyzer.output.pdfalyzer_presenter import PdfalyzerPresenter from pdfalyzer.output.styles.rich_theme import PDFALYZER_THEME_DICT from pdfalyzer.pdfalyzer import Pdfalyzer @@ -41,8 +43,12 @@ def pdfalyze(): args = parse_arguments() pdfalyzer = Pdfalyzer(args.file_to_scan_path) - pdfalyzer = PdfalyzerPresenter(pdfalyzer) + presenter = PdfalyzerPresenter(pdfalyzer) output_basepath = None + + # Initialize JSON exporter if needed + json_exporter = JsonExporter(pdfalyzer) if args.json else None + json_export_data = {} if args.json else None # Binary stream extraction is a special case if args.extract_binary_streams: @@ -53,7 +59,7 @@ def pdfalyze(): # The method that gets called is related to the argument name. See 'possible_output_sections' list in argument_parser.py # Analysis exports wrap themselves around the methods that actually generate the analyses - for (arg, method) in output_sections(args, pdfalyzer): + for (arg, method) in output_sections(args, presenter): if args.output_dir: output_basepath = PdfalyzerConfig.get_output_basepath(method) print(f'Exporting {arg} data to {output_basepath}...') @@ -69,10 +75,52 @@ def pdfalyze(): if args.export_svg: invoke_rich_export(console.save_svg, output_basepath) + + # Handle JSON export + if args.json and json_exporter: + json_output_path = None + if arg == 'docinfo': + json_output_path = json_exporter.export_document_info(Path(args.output_dir)) + elif arg == 'tree': + json_output_path = json_exporter.export_tree(Path(args.output_dir)) + elif arg == 'rich': + # For rich tree, export the detailed tree structure + json_output_path = json_exporter.export_tree(Path(args.output_dir)) + # Rename file to indicate it's the rich/detailed version + if json_output_path.exists(): + rich_path = json_output_path.parent / json_output_path.name.replace('_tree.json', '_rich_tree.json') + json_output_path.rename(rich_path) + json_output_path = rich_path + elif arg == 'counts': + json_output_path = json_exporter.export_summary(Path(args.output_dir)) + elif arg == 'fonts': + json_output_path = json_exporter.export_fonts(Path(args.output_dir)) + elif arg == 'streams': + json_output_path = json_exporter.export_streams(Path(args.output_dir)) + elif arg == 'yara': + # Get YARA matches from the presenter's yaralyzer + if hasattr(presenter, 'yaralyzer') and hasattr(presenter.yaralyzer, 'yara_matches'): + json_output_path = json_exporter.export_yara_results(Path(args.output_dir), presenter.yaralyzer.yara_matches) + + if json_output_path: + json_export_data[arg] = str(json_output_path) + log_and_print(f" -> Exported {arg} to JSON: {json_output_path}") # Clear the buffer if we have one if args.output_dir: del console._record_buffer[:] + + # If JSON export was requested, create a manifest file + if args.json and json_export_data: + manifest_path = Path(args.output_dir) / f"{pdfalyzer.pdf_basename}_manifest.json" + with open(manifest_path, 'w') as f: + import json + json.dump({ + "pdf_file": pdfalyzer.pdf_basename, + "exports": json_export_data, + "timestamp": datetime.now().isoformat() + }, f, indent=2) + log_and_print(f"\nJSON export complete. Manifest written to: {manifest_path}") # Drop into interactive shell if requested if args.interact: diff --git a/pdfalyzer/output/json_exporter.py b/pdfalyzer/output/json_exporter.py new file mode 100644 index 00000000..d14dcebe --- /dev/null +++ b/pdfalyzer/output/json_exporter.py @@ -0,0 +1,344 @@ +""" +JSON export functionality for pdfalyzer. +Serializes PDF analysis results to JSON format while preserving PDF structure. +""" +import json +from pathlib import Path +from typing import Any, Dict, List, Optional, Union +from datetime import datetime + +from anytree import LevelOrderIter, SymlinkNode +from pypdf.generic import ( + ArrayObject, BooleanObject, DictionaryObject, FloatObject, + IndirectObject, NameObject, NullObject, NumberObject, + StreamObject, TextStringObject +) + +from pdfalyzer.decorators.pdf_tree_node import PdfTreeNode +from pdfalyzer.pdfalyzer import Pdfalyzer +from yaralyzer.util.logging import log + + +class PdfJsonEncoder(json.JSONEncoder): + """Custom JSON encoder for PDF objects.""" + + def default(self, obj): + # Handle PDF-specific types + if isinstance(obj, (NameObject, TextStringObject)): + return str(obj) + elif isinstance(obj, (NumberObject, FloatObject)): + return float(obj) + elif isinstance(obj, BooleanObject): + return bool(obj) + elif isinstance(obj, NullObject): + return None + elif isinstance(obj, IndirectObject): + return { + "_type": "IndirectObject", + "idnum": obj.idnum, + "generation": obj.generation + } + elif isinstance(obj, ArrayObject): + return list(obj) + elif isinstance(obj, DictionaryObject): + return {str(k): v for k, v in obj.items()} + elif isinstance(obj, StreamObject): + return { + "_type": "StreamObject", + "dictionary": {str(k): v for k, v in obj.items()}, + "length": len(obj._data) if hasattr(obj, '_data') else 0 + } + elif isinstance(obj, bytes): + # For binary data, encode as base64 or just indicate presence + return { + "_type": "bytes", + "length": len(obj), + "preview": str(obj[:100]) if len(obj) <= 100 else str(obj[:100]) + "..." + } + elif hasattr(obj, '__dict__'): + # For custom objects, try to serialize their attributes + return {k: v for k, v in obj.__dict__.items() if not k.startswith('_')} + else: + return super().default(obj) + + +class JsonExporter: + """Handles JSON export of PDF analysis results.""" + + def __init__(self, pdfalyzer: Pdfalyzer): + self.pdfalyzer = pdfalyzer + self.encoder = PdfJsonEncoder(indent=2) + + def export_all(self, output_dir: Path) -> Dict[str, Path]: + """Export all analysis results to JSON files.""" + output_dir = Path(output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + exported_files = {} + + # Export each analysis type + exported_files['docinfo'] = self.export_document_info(output_dir) + exported_files['tree'] = self.export_tree(output_dir) + exported_files['summary'] = self.export_summary(output_dir) + exported_files['fonts'] = self.export_fonts(output_dir) + exported_files['streams'] = self.export_streams(output_dir) + + return exported_files + + def export_document_info(self, output_dir: Path) -> Path: + """Export document metadata and hashes to JSON.""" + data = { + "pdf_file": self.pdfalyzer.pdf_basename, + "pdf_path": str(self.pdfalyzer.pdf_path), + "metadata": self._serialize_metadata(), + "hashes": { + "md5": self.pdfalyzer.pdf_bytes_info.md5, + "sha1": self.pdfalyzer.pdf_bytes_info.sha1, + "sha256": self.pdfalyzer.pdf_bytes_info.sha256 + }, + "file_size": self.pdfalyzer.pdf_bytes_info.size, + "pdf_version": str(self.pdfalyzer.pdf_reader.pdf_header) if hasattr(self.pdfalyzer.pdf_reader, 'pdf_header') else None, + "max_generation": self.pdfalyzer.max_generation, + "analysis_timestamp": datetime.now().isoformat() + } + + output_path = output_dir / f"{self.pdfalyzer.pdf_basename}_docinfo.json" + with open(output_path, 'w') as f: + json.dump(data, f, indent=2, cls=PdfJsonEncoder) + + return output_path + + def export_tree(self, output_dir: Path) -> Path: + """Export PDF tree structure to JSON.""" + data = { + "pdf_file": self.pdfalyzer.pdf_basename, + "root": self._serialize_node(self.pdfalyzer.pdf_tree), + "total_nodes": len(list(LevelOrderIter(self.pdfalyzer.pdf_tree))) + } + + output_path = output_dir / f"{self.pdfalyzer.pdf_basename}_tree.json" + with open(output_path, 'w') as f: + json.dump(data, f, indent=2, cls=PdfJsonEncoder) + + return output_path + + def export_summary(self, output_dir: Path) -> Path: + """Export node type counts and statistics to JSON.""" + data = self._analyze_tree() + data["pdf_file"] = self.pdfalyzer.pdf_basename + + output_path = output_dir / f"{self.pdfalyzer.pdf_basename}_summary.json" + with open(output_path, 'w') as f: + json.dump(data, f, indent=2, sort_keys=True) + + return output_path + + def export_fonts(self, output_dir: Path) -> Path: + """Export font information to JSON.""" + fonts_data = [] + + for font_info in self.pdfalyzer.font_infos: + font_data = { + "idnum": font_info.idnum if hasattr(font_info, 'idnum') else None, + "label": font_info.label if hasattr(font_info, 'label') else None, + "type": str(font_info.font_obj.get('/Type', '')) if hasattr(font_info, 'font_obj') else None, + "subtype": str(font_info.font_obj.get('/Subtype', '')) if hasattr(font_info, 'font_obj') else None, + "base_font": str(font_info.font_obj.get('/BaseFont', '')) if hasattr(font_info, 'font_obj') else None, + "encoding": str(font_info.font_obj.get('/Encoding', '')) if hasattr(font_info, 'font_obj') else None, + "has_font_file": font_info.font_file is not None if hasattr(font_info, 'font_file') else False, + "has_to_unicode": font_info.font_obj.get('/ToUnicode') is not None if hasattr(font_info, 'font_obj') else False + } + + # Add character mapping info if available + if hasattr(font_info, 'character_mapping') and font_info.character_mapping: + font_data["character_count"] = len(font_info.character_mapping) + font_data["character_mapping_sample"] = dict(list(font_info.character_mapping.items())[:10]) + + fonts_data.append(font_data) + + data = { + "pdf_file": self.pdfalyzer.pdf_basename, + "total_fonts": len(fonts_data), + "fonts": fonts_data + } + + output_path = output_dir / f"{self.pdfalyzer.pdf_basename}_fonts.json" + with open(output_path, 'w') as f: + json.dump(data, f, indent=2) + + return output_path + + def export_streams(self, output_dir: Path, include_data: bool = False) -> Path: + """Export stream information to JSON.""" + streams_data = [] + + for node in self.pdfalyzer.stream_nodes(): + stream_data = { + "idnum": node.idnum, + "generation": node.obj.generation if hasattr(node.obj, 'generation') else 0, + "type": str(node.type), + "subtype": str(node.sub_type) if node.sub_type else None, + "stream_length": node.stream_length, + "filters": self._get_stream_filters(node), + "first_address": node.first_address + } + + # Optionally include stream data (be careful with large PDFs) + if include_data and node.stream_data: + if len(node.stream_data) <= 1000: # Only include small streams + stream_data["data_preview"] = str(node.stream_data[:500]) + else: + stream_data["data_preview"] = f"Stream too large ({len(node.stream_data)} bytes)" + + streams_data.append(stream_data) + + data = { + "pdf_file": self.pdfalyzer.pdf_basename, + "total_streams": len(streams_data), + "streams": streams_data + } + + output_path = output_dir / f"{self.pdfalyzer.pdf_basename}_streams.json" + with open(output_path, 'w') as f: + json.dump(data, f, indent=2) + + return output_path + + def export_yara_results(self, output_dir: Path, yara_matches: List[Any]) -> Path: + """Export YARA scan results to JSON.""" + matches_data = [] + + for match in yara_matches: + match_data = { + "rule": match.rule, + "namespace": match.namespace, + "tags": list(match.tags) if match.tags else [], + "meta": dict(match.meta) if match.meta else {}, + "strings": [] + } + + for string in match.strings: + string_data = { + "offset": string[0], + "identifier": string[1], + "string": str(string[2]) + } + match_data["strings"].append(string_data) + + matches_data.append(match_data) + + data = { + "pdf_file": self.pdfalyzer.pdf_basename, + "total_matches": len(matches_data), + "matches": matches_data + } + + output_path = output_dir / f"{self.pdfalyzer.pdf_basename}_yara.json" + with open(output_path, 'w') as f: + json.dump(data, f, indent=2) + + return output_path + + def _serialize_node(self, node: PdfTreeNode) -> Dict[str, Any]: + """Serialize a PDF tree node to a dictionary.""" + if isinstance(node, SymlinkNode): + return { + "_type": "SymlinkNode", + "target_idnum": node.target.idnum if hasattr(node.target, 'idnum') else None, + "reference_key": str(node.reference_key) if hasattr(node, 'reference_key') else None + } + + node_data = { + "idnum": node.idnum, + "type": str(node.type), + "subtype": str(node.sub_type) if node.sub_type else None, + "label": node.label, + "first_address": node.first_address, + "known_to_parent_as": node.known_to_parent_as, + "generation": node.obj.generation if hasattr(node.obj, 'generation') else 0, + "has_stream": node.stream_data is not None if hasattr(node, 'stream_data') else False, + "stream_length": node.stream_length if hasattr(node, 'stream_length') and node.stream_length > 0 else None + } + + # Add PDF object data (simplified) + if node.obj: + try: + if isinstance(node.obj, DictionaryObject): + node_data["object_data"] = {str(k): v for k, v in node.obj.items()} + elif isinstance(node.obj, ArrayObject): + node_data["object_data"] = list(node.obj) + else: + node_data["object_data"] = str(node.obj) + except Exception as e: + node_data["object_data"] = f"Error serializing object: {str(e)}" + + # Add children + if node.children: + node_data["children"] = [self._serialize_node(child) for child in node.children] + + # Add non-tree relationships + if node.non_tree_relationships: + node_data["references"] = [ + { + "to_idnum": rel.to_obj.idnum if hasattr(rel.to_obj, 'idnum') else None, + "reference_key": str(rel.reference_key), + "is_link": rel.is_link + } + for rel in node.non_tree_relationships + ] + + return node_data + + def _serialize_metadata(self) -> Dict[str, Any]: + """Serialize PDF metadata.""" + metadata = {} + if self.pdfalyzer.pdf_reader.metadata: + for key, value in self.pdfalyzer.pdf_reader.metadata.items(): + key_str = str(key) + if hasattr(value, 'strftime'): # It's a datetime + metadata[key_str] = value.isoformat() + else: + metadata[key_str] = str(value) + return metadata + + def _analyze_tree(self) -> Dict[str, Any]: + """Analyze tree structure and return statistics.""" + node_type_counts = {} + node_subtype_counts = {} + total_stream_bytes = 0 + + for node in LevelOrderIter(self.pdfalyzer.pdf_tree): + if isinstance(node, SymlinkNode): + continue + + # Count by type + type_str = str(node.type) + node_type_counts[type_str] = node_type_counts.get(type_str, 0) + 1 + + # Count by subtype + if node.sub_type: + subtype_str = str(node.sub_type) + node_subtype_counts[subtype_str] = node_subtype_counts.get(subtype_str, 0) + 1 + + # Sum stream bytes + if hasattr(node, 'stream_length') and node.stream_length and node.stream_length > 0: + total_stream_bytes += node.stream_length + + return { + "total_objects": len(list(LevelOrderIter(self.pdfalyzer.pdf_tree))), + "node_type_counts": node_type_counts, + "node_subtype_counts": node_subtype_counts, + "total_stream_bytes": total_stream_bytes, + "max_tree_depth": self.pdfalyzer.pdf_tree.height + } + + def _get_stream_filters(self, node: PdfTreeNode) -> List[str]: + """Extract filter information from a stream node.""" + filters = [] + if hasattr(node.obj, 'get') and '/Filter' in node.obj: + filter_obj = node.obj['/Filter'] + if isinstance(filter_obj, ArrayObject): + filters = [str(f) for f in filter_obj] + else: + filters = [str(filter_obj)] + return filters \ No newline at end of file diff --git a/pdfalyzer/util/argument_parser.py b/pdfalyzer/util/argument_parser.py index 4315aecc..36c7e35d 100644 --- a/pdfalyzer/util/argument_parser.py +++ b/pdfalyzer/util/argument_parser.py @@ -50,6 +50,10 @@ const='bin', help='extract all binary streams in the PDF to separate files (requires pdf-parser.py)') +export.add_argument('-json', + action='store_true', + help='export analysis results to JSON files preserving PDF structure') + # Add one more option to the YARA rules section source.add_argument('--no-default-yara-rules', action='store_true', @@ -140,7 +144,7 @@ def parse_arguments(): exit_with_error("--no-default-yara-rules requires at least one --yara-file argument") # File export options - if args.export_svg or args.export_txt or args.export_html or args.extract_binary_streams: + if args.export_svg or args.export_txt or args.export_html or args.json or args.extract_binary_streams: args.output_dir = args.output_dir or getcwd() file_prefix = (args.file_prefix + '__') if args.file_prefix else '' args.file_suffix = ('_' + args.file_suffix) if args.file_suffix else '' diff --git a/tests/conftest.py b/tests/conftest.py index d7513be8..ff1b4264 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -27,6 +27,10 @@ def adobe_type1_fonts_pdf_path(): def analyzing_malicious_pdf_path(): return _pdf_in_doc_dir('analyzing-malicious-document-files.pdf') +@pytest.fixture(scope='session') +def one_page_pdf_path(): + return str(FIXTURES_DIR / 'one_page_pdfs' / 'page_1.pdf') + # Some obj ids for use with -f when you want to limit yourself to the font @pytest.fixture(scope="session") diff --git a/tests/test_file_export.py b/tests/test_file_export.py index 07bddba0..58ccda0c 100644 --- a/tests/test_file_export.py +++ b/tests/test_file_export.py @@ -1,3 +1,4 @@ +import json from math import isclose from os import environ, path, remove from subprocess import check_output @@ -33,3 +34,61 @@ def assert_array_is_close(_list1, _list2): for i, item in enumerate(_list1): if not isclose(item, _list2[i], rel_tol=0.05): assert False, f"File size of {item} too far from {_list2[i]}" + + +def test_json_export(one_page_pdf_path, tmp_dir): + """Test JSON export functionality.""" + import sys + args = [ + sys.executable, '-m', 'pdfalyzer', + one_page_pdf_path, + '-json', + '--output-dir', tmp_dir, + '--docinfo', + '--tree', + '--counts', + '--fonts' + ] + + check_output(args, env=environ) + + # Check that JSON files were created + json_files = [f for f in files_in_dir(tmp_dir) if f.endswith('.json')] + assert len(json_files) >= 4 # At least docinfo, tree, summary, and manifest + + # Verify each JSON file is valid and contains expected data + for json_file in json_files: + with open(json_file, 'r') as f: + data = json.load(f) + assert isinstance(data, dict) + + # Check specific file contents + if 'docinfo.json' in json_file: + assert 'metadata' in data + assert 'hashes' in data + assert 'file_size' in data + assert 'pdf_file' in data + + elif 'tree.json' in json_file: + assert 'root' in data + assert 'total_nodes' in data + assert 'pdf_file' in data + + elif 'summary.json' in json_file: + assert 'node_type_counts' in data + assert 'total_objects' in data + assert 'pdf_file' in data + + elif 'manifest.json' in json_file: + assert 'exports' in data + assert 'timestamp' in data + assert 'pdf_file' in data + + elif 'fonts.json' in json_file: + assert 'fonts' in data + assert 'total_fonts' in data + assert 'pdf_file' in data + + # Clean up + for file in json_files: + remove(file)