Read, decrypt, and parse JSON data from a GPG-encrypted JSON file.

gpg_json_reader.py

#!/usr/bin/env python3

import json
import os
import shutil
import subprocess
from pathlib import Path


class GPGJSONReaderError(Exception):
    """
    Custom exception class for errors occurring during GPG JSON reading.
    """
    pass


class GPGJSONReader:
    """
    Decodes and parses a GPG-encrypted JSON file.

    This class provides a mechanism to read, decrypt, and parse JSON data from
    a GPG-encrypted file. The decryption process leverages the system's GPG
    binary. Use this class to handle sensitive JSON data securely.

    Attributes:
    GPG_BINARY (str): Path to the GPG binary on the system.
    GPG_FLAGS (list[str]): Flags used for the GPG decryption command.
    TIMEOUT (int): Timeout duration (in seconds) for the GPG decryption process.
    """
    GPG_BINARY = shutil.which('gpg')
    GPG_FLAGS = ['--quiet', '--batch', '--decrypt']
    TIMEOUT = 30  # seconds

    def __init__(self, json_gpg_file: Path) -> None:
        self.json_gpg_file = Path(json_gpg_file)  # Make sure it's a Path object.

    def _validate_input_file(self) -> None:
        if not self.json_gpg_file.exists():
            raise GPGJSONReaderError(f"File does not exist: {self.json_gpg_file}")
        if not self.json_gpg_file.is_file():
            raise GPGJSONReaderError(f"Path is not a file: {self.json_gpg_file}")
        if not os.access(self.json_gpg_file, os.R_OK):
            raise GPGJSONReaderError(f"File is not readable: {self.json_gpg_file}")

    def remove_instance_attribute(self, attr_name: str) -> None:
        """
        Removes an instance attribute if it exists.

        Args:
            attr_name: The name of the attribute to remove.

        Note:
            Silently ignores if the attribute doesn't exist.
        """
        if hasattr(self, attr_name):
            delattr(self, attr_name)

    def _decrypt_gpg_file(self) -> str:
        try:
            decrypted_data = subprocess.run(  # Decrypted JSON data. Remember to clean up later.
                [self.GPG_BINARY, *self.GPG_FLAGS, str(self.json_gpg_file)],
                check=True,
                capture_output=True,
                text=True,
                timeout=self.TIMEOUT
            )
            if decrypted_data.stderr:
                raise GPGJSONReaderError(f"GPG warning during decryption: {decrypted_data.stderr}")
            return decrypted_data.stdout

        except subprocess.TimeoutExpired:
            raise GPGJSONReaderError(f"GPG decrypted_data timed out after {self.TIMEOUT} seconds")
        except subprocess.CalledProcessError as e:
            raise GPGJSONReaderError(f"GPG decryption failed: {e.stderr}")

    def _parse_json_data(self):
        try:
            return json.loads(self._decrypt_gpg_file())
        except json.JSONDecodeError as e:
            raise GPGJSONReaderError(f"Failed to parse decrypted JSON data: {e}")

    def get_parsed_json(self):
        try:
            self._validate_input_file()  # Preparatory checks.
            return self._parse_json_data()
        finally:
            # Clean up sensitive data
            sensitive_data = ['json_gpg_file', 'decrypted_data']
            for attr in sensitive_data:
                self.remove_instance_attribute(attr)


if __name__ == "__main__":
    # Example usage:
    reader = GPGJSONReader(Path(input("Enter the path to the JSON file: ")))
    try:
        print(reader.get_parsed_json())
    except GPGJSONReaderError as exception:
        print(f"Error processing file: {exception}")