From 5b56692b023d3e00e6e539b740b5e8bdce070a46 Mon Sep 17 00:00:00 2001 From: Sandro <34354689+Lainupcomputer@users.noreply.github.com> Date: Fri, 18 Sep 2026 04:52:12 +0200 Subject: [PATCH 1/2] Rename project to PyLockbox --- .github/workflows/ci.yml | 25 + .gitignore | 27 +- LICENSE | 21 + README.md | 174 +++- pyproject.toml | 64 +- scripts/publish.ps1 | 12 + scripts/publish.py | 69 ++ src/ez_storage/ez_storage.py | 182 ---- src/pylockbox/__init__.py | 32 + src/pylockbox/errors.py | 39 + .../__init__.py => pylockbox/py.typed} | 0 src/pylockbox/secure_store.py | 948 ++++++++++++++++++ tests/TEST.py | 45 - tests/test_secure_store.py | 160 +++ 14 files changed, 1558 insertions(+), 240 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 LICENSE create mode 100644 scripts/publish.ps1 create mode 100644 scripts/publish.py delete mode 100644 src/ez_storage/ez_storage.py create mode 100644 src/pylockbox/__init__.py create mode 100644 src/pylockbox/errors.py rename src/{ez_storage/__init__.py => pylockbox/py.typed} (100%) create mode 100644 src/pylockbox/secure_store.py delete mode 100644 tests/TEST.py create mode 100644 tests/test_secure_store.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..d2e7ebd --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,25 @@ +name: CI + +on: + push: + pull_request: + +jobs: + tests: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.10", "3.11", "3.12", "3.13"] + + steps: + - name: Check out repository + uses: actions/checkout@v4 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install package + run: python -m pip install . + - name: Run tests + run: python -m unittest discover -s tests -v diff --git a/.gitignore b/.gitignore index 380d773..b8d1137 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,25 @@ -/dist/ -/src/ +__pycache__/ +*.py[cod] + +.venv/ +venv/ +ENV/ + +build/ +dist/ +*.egg-info/ + +.pytest_cache/ +.ruff_cache/ +.mypy_cache/ +.coverage +htmlcov/ + +*.log +*.tmp +*.lock +*.lockbox +*.lockbox.bak.* + +.env +.env.local diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..1ac01f5 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Sandro Kalett + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index ef5a4f5..956683f 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,172 @@ -# EZ Storage +# PyLockbox -mostly it is a dependency +`PyLockbox` is a small general-purpose object store for Python applications. It stores one +Pickle object in an encrypted, authenticated binary file and provides a simple key/value API for +the common dictionary case. -function overview provided in `/tests/TEST.py` +Version 2 is a breaking rewrite. It intentionally does not read the old JSON format and does not +include a legacy migration layer. -## Installation: +## Install -` python -m pip install ez-storage` -## Usages: +```bash +python -m pip install pylockbox +``` -1. used in [stable-diffusion-webui-random_prompt_generator](https://github.com/Lainupcomputer/stable-diffusion-webui-random_prompt_generator) +For development: +```bash +python -m pip install -e ".[dev]" +``` -## Notes: -feel free to use \ No newline at end of file +## Quick start + +The key must be supplied as raw 32-byte data or as base64/hex text. There is no default key. + +```python +from pylockbox import SecureStore, encode_key, generate_key + +key = generate_key() +print("Store this outside your repository:", encode_key(key)) + +store = SecureStore("state.lockbox", key=key) +store.set("user.name", "Lainup") +store.set("settings.theme", "dark") +store.save() + +reader = SecureStore("state.lockbox", key=key) +print(reader.load()) +print(reader.get("user.name")) +``` + +`save(value)` and `load()` also work with any Pickle-compatible root value. The mapping helpers +(`set`, `get`, `remove`, `keys`, and `items`) require the root value to be a dictionary. Dotted +names such as `settings.theme` address nested dictionaries. + +## Environment variables + +Use `SecureStore.from_env()` to load configuration from environment variables. The default prefix +is `PYLOCKBOX_`; a custom prefix is supported for applications with multiple stores. + +| Variable | Default | Meaning | +| --- | --- | --- | +| `PYLOCKBOX_PATH` | `storage.lockbox` | Storage file path | +| `PYLOCKBOX_KEY` | required | URL-safe base64 or hex encoded 32-byte key | +| `PYLOCKBOX_MAX_FILE_SIZE` | `64MiB` | Maximum file and decoded payload size | +| `PYLOCKBOX_BACKUPS` | `3` | Number of rotating backups; `0` disables backups | +| `PYLOCKBOX_LOCK_TIMEOUT` | `5` | Lock wait time in seconds | +| `PYLOCKBOX_COMPRESSION` | `true` | Compress before encryption when it saves space | + +Example in PowerShell: + +```powershell +$env:PYLOCKBOX_PATH = "state.lockbox" +$env:PYLOCKBOX_KEY = "paste-a-generated-base64-key-here" +$env:PYLOCKBOX_MAX_FILE_SIZE = "128MiB" +$env:PYLOCKBOX_BACKUPS = "5" +``` + +```python +from pylockbox import SecureStore + +store = SecureStore.from_env() +store.set("counter", 1).save() +``` + +Configuration precedence is: + +```text +per-call override > constructor argument > environment variable > default +``` + +For example, a one-time load override does not change the store configuration: + +```python +value = store.load(key=another_key, max_file_size="256MiB") +``` + +The library reads environment variables; it never writes secrets back into `os.environ`. + +## Security model + +Each file contains a versioned binary envelope, not JSON: + +- ChaCha20-Poly1305 encrypts and authenticates the payload. +- HKDF separates the encryption and HMAC keys derived from the supplied 32-byte key. +- HMAC-SHA256 authenticates the header and ciphertext as an additional integrity layer. +- Pickle protocol 4/5 opcodes are checked before loading. +- A restricted unpickler allows safe built-in values and classes explicitly registered in code. +- No module is imported because a file requests it; unknown globals and persistent IDs are rejected. +- Maximum file size is checked before reading, and decompression is bounded. +- Writes use a temporary file, `fsync`, and atomic replacement. A cross-platform file lock protects + concurrent readers/writers, and POSIX files are created with mode `0600` where supported. + +### Important Pickle limitation + +Pickle is not a security sandbox. This package makes loading substantially safer for authenticated +application state, but no implementation can make arbitrary Pickle data completely safe if an +attacker controls the Python process, obtains the key, or is allowed to register a malicious class. +Only load files protected by a key you trust, and register only classes whose deserialization code +you trust. + +## User-defined classes + +Custom classes are denied by default. Register the exact class on every process that loads the +file: + +```python +from dataclasses import dataclass +from pylockbox import SecureStore + +@dataclass +class Profile: + name: str + +key = SecureStore.generate_key() +writer = SecureStore("profiles.lockbox", key=key) +writer.register_type(Profile) +writer.save(Profile("Lainup")) + +reader = SecureStore("profiles.lockbox", key=key) +reader.register_type(Profile) +profile = reader.load() +``` + +This explicit registration is deliberate: arbitrary functions, modules, and classes are not loaded +from file data. + +## Development and publishing + +Run the tests with the standard library: + +```bash +python -m unittest discover -s tests -v +``` + +The repository includes a build/check/upload script. It does not upload unless `--publish` is +passed, and it never contains a PyPI token: + +```bash +# Build and validate only +python scripts/publish.py --clean + +# Recommended first upload +python scripts/publish.py --repository testpypi --publish + +# Public PyPI upload after TestPyPI verification +python scripts/publish.py --repository pypi --publish +``` + +On Windows PowerShell the wrapper can be used instead: + +```powershell +.\scripts\publish.ps1 --clean +.\scripts\publish.ps1 --repository testpypi --publish +``` + +Configure credentials through Twine's normal mechanisms (`TWINE_USERNAME`, `TWINE_PASSWORD`, a +keyring, or `.pypirc`). Never commit credentials or a generated storage key. + +## License + +MIT. See [LICENSE](LICENSE). diff --git a/pyproject.toml b/pyproject.toml index b5a3c46..9ebb367 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,66 @@ [build-system] requires = [ - "setuptools>=42", + "setuptools>=77", "wheel" ] -build-backend = "setuptools.build_meta" \ No newline at end of file +build-backend = "setuptools.build_meta" + +[project] +name = "pylockbox" +version = "2.0.0" +description = "Encrypted, authenticated and restricted Pickle storage for Python applications" +readme = { file = "README.md", content-type = "text/markdown" } +requires-python = ">=3.10" +license = "MIT" +license-files = ["LICENSE"] +authors = [ + { name = "Sandro Kalett", email = "kontakt@lainupcomputersolution.de" } +] +keywords = ["storage", "pickle", "encryption", "security", "persistence"] +classifiers = [ + "Development Status :: 4 - Beta", + "Intended Audience :: Developers", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3 :: Only", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", + "Topic :: Security :: Cryptography", + "Topic :: Software Development :: Libraries :: Python Modules", +] +dependencies = [ + "cryptography>=42", +] + +[project.urls] +Homepage = "https://github.com/Lainupcomputer/PyLockbox" +Repository = "https://github.com/Lainupcomputer/PyLockbox" +Issues = "https://github.com/Lainupcomputer/PyLockbox/issues" + +[project.optional-dependencies] +dev = [ + "build>=1.2", + "pytest>=8", + "ruff>=0.8", + "twine>=5", +] + +[tool.setuptools] +package-dir = { "" = "src" } +include-package-data = true + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.setuptools.package-data] +pylockbox = ["py.typed"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra" + +[tool.ruff] +line-length = 100 +target-version = "py310" diff --git a/scripts/publish.ps1 b/scripts/publish.ps1 new file mode 100644 index 0000000..a754dcf --- /dev/null +++ b/scripts/publish.ps1 @@ -0,0 +1,12 @@ +[CmdletBinding()] +param( + [Parameter(ValueFromRemainingArguments = $true)] + [string[]]$Arguments +) + +$ErrorActionPreference = "Stop" +$script = Join-Path $PSScriptRoot "publish.py" +& python $script @Arguments +if ($LASTEXITCODE -ne 0) { + exit $LASTEXITCODE +} diff --git a/scripts/publish.py b/scripts/publish.py new file mode 100644 index 0000000..64d5452 --- /dev/null +++ b/scripts/publish.py @@ -0,0 +1,69 @@ +#!/usr/bin/env python3 +"""Build, validate and optionally publish PyLockbox to PyPI or TestPyPI.""" + +from __future__ import annotations + +import argparse +import shutil +import subprocess +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] + + +def run(*args: str) -> None: + command = [sys.executable, *args] + print("+", " ".join(command)) + subprocess.run(command, cwd=ROOT, check=True) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--repository", + choices=("testpypi", "pypi"), + default="testpypi", + help="upload target when --publish is used (default: testpypi)", + ) + parser.add_argument("--publish", action="store_true", help="upload artifacts after validation") + parser.add_argument("--skip-tests", action="store_true", help="skip the local unittest suite") + parser.add_argument("--clean", action="store_true", help="remove build artifacts before building") + args = parser.parse_args() + + if args.clean: + for directory_name in ("build", "dist"): + directory = ROOT / directory_name + if directory.exists(): + shutil.rmtree(directory) + for metadata_directory in ROOT.glob("*.egg-info"): + if metadata_directory.is_dir(): + shutil.rmtree(metadata_directory) + + if not args.skip_tests: + run("-m", "unittest", "discover", "-s", "tests", "-v") + + run("-m", "build") + artifacts = sorted( + path + for path in (ROOT / "dist").iterdir() + if path.is_file() and (path.name.endswith(".whl") or path.name.endswith(".tar.gz")) + ) + if not artifacts: + raise RuntimeError("no wheel or source archive was created in dist/") + + run("-m", "twine", "check", *(str(path) for path in artifacts)) + if args.publish: + upload_args = ["-m", "twine", "upload"] + if args.repository == "testpypi": + upload_args.extend(("--repository", "testpypi")) + upload_args.extend(str(path) for path in artifacts) + run(*upload_args) + else: + print("Build and validation completed. Use --publish to upload the artifacts.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/ez_storage/ez_storage.py b/src/ez_storage/ez_storage.py deleted file mode 100644 index db2f20f..0000000 --- a/src/ez_storage/ez_storage.py +++ /dev/null @@ -1,182 +0,0 @@ -import json -import logging - - -logger = logging.getLogger() -_version = "1.3.3" - - -class Ez_Storage: - def __init__(self, file_path=str("config")): - self.enable_debug = True - self.enable_logging = True - self.storage_prefix = "ez_storage" - self.on_file_creation_placeholder = "{\n\n}" - self.internal_config = None - - self.filepath = file_path - if self.check_file() == "data createddata found": - self.restore_storage("b") - - @staticmethod - def read_file(file_path): - with open(file_path, "r") as f: - file = json.load(f) - return file - - @staticmethod - def save(file_path, file): - with open(file_path, "w") as f: - json.dump(file, f, indent=2) - - def make_response(self, level, response): - if self.enable_logging: - logger.log(level, response) - elif self.enable_debug: - print(f"LEVEL:{level},msg:{response}") - else: - pass - - def check_file(self): - file_found = False - result = "" - while not file_found: - try: - self.read_file(self.filepath) - result += "data found" - file_found = True - - except FileNotFoundError: - with open(self.filepath, "w") as f: - f.write(self.on_file_creation_placeholder) - result += "data created" - file_found = False - - self.make_response(logging.INFO, f"[{self.storage_prefix}]:{result}") - return result - - def add_storage(self, mode=str(), obj=None, data=None, value=None, array_data=None, override=False): - if mode == "o": - file = self.read_file(self.filepath) - try: - if override: - file[obj][data] = value - result = f"{obj}, {data}, {value} override" - else: - result = f"{obj}, {data}, {value} override protected" - except KeyError: - file[obj] = {} - file[obj][data] = value - result = f"'{obj}, {data}, {value}' added as object" - - self.save(self.filepath, file) - - elif mode == "a": - file = self.read_file(self.filepath) - try: - if override: - file[obj] = [] - file[obj].append(array_data) - result = f"{obj}, {array_data} override" - else: - file[obj].append(array_data) - result = f"{obj}, {array_data} appended" - except KeyError: - file[obj] = [] - file[obj].append(array_data) - result = f"'{obj}, {array_data}' added as array" - - self.save(self.filepath, file) - - elif mode == "l": - file = self.read_file(self.filepath) - try: - if override: - file[obj].clear() - for x in data: - file[obj].append(x) - result = f"{obj}, {data} override" - else: - for x in data: - file[obj].append(x) - result = f"{obj}, {data} appended" - except KeyError: - file[obj] = [] - for x in data: - file[obj].append(x) - result = f"'{obj}, {data}' added as list" - - self.save(self.filepath, file) - - else: - result = f"Error: No defined for: mode'{mode}'" - - self.make_response(logging.INFO, f"[{self.storage_prefix}]:{result}") - - def get_storage(self, mode=str(), obj=None, data=None): - file = self.read_file(self.filepath) - if mode == "o": - try: - r_data = file[obj][data] - self.make_response(logging.INFO, f"[{self.storage_prefix}]:get {obj} @ {data} in object mode: Success") - except KeyError: - r_data = None - self.make_response(logging.INFO, f"[{self.storage_prefix}]:get {obj} @ {data} in object mode: KeyError") - - return r_data - - elif mode == "a": - objects = [] - for file_obj in file[obj]: - objects.append(file_obj) - self.make_response(logging.INFO, f"[{self.storage_prefix}]:get {obj} in array mode: Success") - return objects - - elif mode == "l": - objects = [] - for file_obj in file[obj]: - objects.append(file_obj) - self.make_response(logging.INFO, f"[{self.storage_prefix}]:get {obj} in list mode: Success") - return objects - - else: - self.make_response(logging.INFO, f"[{self.storage_prefix}]:Error: No defined for: mode'{mode}'") - - def restore_storage(self, source=None, destination=None, mode=None): - action_pass = False - result = "" - file = {"ez_storage": {}} - file["ez_storage"]["version"] = _version - - if source == "b": - result = " restored storage form source 'b': Success" - action_pass = True - - elif source == "i": - if mode == "o": - file[destination] = {} - for obj, val in self.internal_config: - file[destination][obj] = val - result = " restored storage form source 'i', mode'o': Success" - action_pass = True - - elif mode == "a": - file[destination] = [] - file[destination].append(self.internal_config) - result = " restored storage form source 'i', mode'a': Success" - action_pass = True - - elif mode == "l": - file[destination] = [] - for x in self.internal_config: - file[destination].append(x) - result = " restored storage form source 'i', mode'l': Success" - action_pass = True - - else: - result = f"Error: No defined mode for: source 'i', mode'{mode}'" - - if action_pass: - self.save(self.filepath, file) - - self.make_response(logging.INFO, f"[{self.storage_prefix}]:{result}") diff --git a/src/pylockbox/__init__.py b/src/pylockbox/__init__.py new file mode 100644 index 0000000..f840544 --- /dev/null +++ b/src/pylockbox/__init__.py @@ -0,0 +1,32 @@ +"""Secure encrypted Pickle storage for Python applications.""" + +from .errors import ( + MissingKeyError, + StorageConfigurationError, + StorageError, + StorageFormatError, + StorageIntegrityError, + StorageLockError, + StorageSizeError, + StorageTypeError, + UnsafeTypeError, +) +from .secure_store import SecureStore, encode_key, generate_key + +__version__ = "2.0.0" + +__all__ = [ + "SecureStore", + "encode_key", + "generate_key", + "MissingKeyError", + "StorageConfigurationError", + "StorageError", + "StorageFormatError", + "StorageIntegrityError", + "StorageLockError", + "StorageSizeError", + "StorageTypeError", + "UnsafeTypeError", + "__version__", +] diff --git a/src/pylockbox/errors.py b/src/pylockbox/errors.py new file mode 100644 index 0000000..d347a29 --- /dev/null +++ b/src/pylockbox/errors.py @@ -0,0 +1,39 @@ +"""Exception types raised by :mod:`pylockbox`.""" + +from __future__ import annotations + + +class StorageError(Exception): + """Base class for all PyLockbox errors.""" + + +class StorageConfigurationError(StorageError, ValueError): + """The store configuration is missing or invalid.""" + + +class MissingKeyError(StorageConfigurationError): + """No encryption key was supplied by code or the environment.""" + + +class StorageFormatError(StorageError): + """The file is not a supported PyLockbox envelope.""" + + +class StorageIntegrityError(StorageError): + """The file failed its authentication or integrity checks.""" + + +class StorageSizeError(StorageError): + """The file or its decoded payload exceeds the configured limit.""" + + +class UnsafeTypeError(StorageError): + """The restricted Pickle loader encountered an unapproved type or opcode.""" + + +class StorageLockError(StorageError): + """The storage file could not be locked before the timeout.""" + + +class StorageTypeError(StorageError, TypeError): + """A key/value operation requires a mapping but found another object.""" diff --git a/src/ez_storage/__init__.py b/src/pylockbox/py.typed similarity index 100% rename from src/ez_storage/__init__.py rename to src/pylockbox/py.typed diff --git a/src/pylockbox/secure_store.py b/src/pylockbox/secure_store.py new file mode 100644 index 0000000..b10e551 --- /dev/null +++ b/src/pylockbox/secure_store.py @@ -0,0 +1,948 @@ +"""Encrypted, authenticated and restricted Pickle-backed object storage. + +The module deliberately does not expose a general-purpose ``pickle.loads`` +path. Files are authenticated before decryption and decoded through an +allowlist-based unpickler with no dynamic imports. +""" + +from __future__ import annotations + +import base64 +import hashlib +import hmac +import io +import logging +import os +import pickle +import pickletools +import secrets +import shutil +import struct +import tempfile +import time +import zlib +from collections.abc import Iterable, Mapping +from contextlib import contextmanager +from pathlib import Path +from re import fullmatch +from typing import Any, ClassVar, Iterator, TypeVar + +from cryptography.exceptions import InvalidTag +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305 +from cryptography.hazmat.primitives.kdf.hkdf import HKDF + +from .errors import ( + MissingKeyError, + StorageConfigurationError, + StorageFormatError, + StorageIntegrityError, + StorageLockError, + StorageSizeError, + StorageTypeError, + UnsafeTypeError, +) + +logger = logging.getLogger(__name__) + +_T = TypeVar("_T") +_MISSING = object() + +KEY_SIZE = 32 +NONCE_SIZE = 12 +HMAC_SIZE = 32 +DEFAULT_MAX_FILE_SIZE = 64 * 1024 * 1024 +DEFAULT_BACKUPS = 3 +DEFAULT_LOCK_TIMEOUT = 5.0 +DEFAULT_COMPRESSION = True +DEFAULT_PATH = "storage.lockbox" + +_MAGIC = b"EZST" +_FORMAT_VERSION = 1 +_FLAG_COMPRESSED = 0x01 +_KNOWN_FLAGS = _FLAG_COMPRESSED +_HEADER = struct.Struct("!4sBBHQQ12s") +_ENVELOPE_OVERHEAD = _HEADER.size + HMAC_SIZE + 16 + +# This is intentionally a small set. Lists, dictionaries, strings, numbers +# and byte values are useful application state and do not require imports or +# user-defined code during loading. Custom classes must be registered on the +# store before loading. +_SAFE_BUILTINS: dict[tuple[str, str], type[Any]] = { + ("builtins", "NoneType"): type(None), + ("builtins", "bool"): bool, + ("builtins", "int"): int, + ("builtins", "float"): float, + ("builtins", "complex"): complex, + ("builtins", "bytes"): bytes, + ("builtins", "bytearray"): bytearray, + ("builtins", "str"): str, + ("builtins", "tuple"): tuple, + ("builtins", "list"): list, + ("builtins", "dict"): dict, + ("builtins", "set"): set, + ("builtins", "frozenset"): frozenset, +} + +# Protocol 4/5 structural opcodes plus the opcodes needed for explicitly +# registered classes. GLOBAL/STACK_GLOBAL are still gated by +# _RestrictedUnpickler.find_class, so they can never import arbitrary names. +_ALLOWED_OPCODES = { + "PROTO", + "FRAME", + "STOP", + "NONE", + "NEWTRUE", + "NEWFALSE", + "BININT", + "BININT1", + "BININT2", + "LONG1", + "LONG4", + "BINFLOAT", + "SHORT_BINUNICODE", + "BINUNICODE", + "BINUNICODE8", + "SHORT_BINBYTES", + "BINBYTES", + "BINBYTES8", + "BYTEARRAY8", + "EMPTY_TUPLE", + "TUPLE", + "TUPLE1", + "TUPLE2", + "TUPLE3", + "EMPTY_LIST", + "LIST", + "APPEND", + "APPENDS", + "EMPTY_DICT", + "DICT", + "SETITEM", + "SETITEMS", + "EMPTY_SET", + "ADDITEMS", + "FROZENSET", + "MARK", + "MEMOIZE", + "BINPUT", + "LONG_BINPUT", + "BINGET", + "LONG_BINGET", + "POP", + "POP_MARK", + "DUP", + "GLOBAL", + "STACK_GLOBAL", + "REDUCE", + "NEWOBJ", + "NEWOBJ_EX", + "BUILD", +} + + +def generate_key() -> bytes: + """Return a cryptographically random 256-bit storage key.""" + + return secrets.token_bytes(KEY_SIZE) + + +def encode_key(key: bytes | bytearray | memoryview) -> str: + """Encode a raw 32-byte key for an environment variable.""" + + raw = _normalise_key(key) + return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") + + +def _normalise_key(value: bytes | bytearray | memoryview | str) -> bytes: + if isinstance(value, (bytes, bytearray, memoryview)): + raw = bytes(value) + elif isinstance(value, str): + text = value.strip() + if text.startswith("base64:"): + text = text[7:] + raw = _decode_base64(text) + elif text.startswith("b64:"): + text = text[4:] + raw = _decode_base64(text) + elif text.startswith("hex:"): + try: + raw = bytes.fromhex(text[4:]) + except ValueError as exc: + raise StorageConfigurationError("EZ storage key is not valid hex") from exc + else: + try: + raw = _decode_base64(text) + except StorageConfigurationError: + try: + raw = bytes.fromhex(text) + except ValueError as exc: + raise StorageConfigurationError( + "key must be 32 raw bytes, base64, or hex" + ) from exc + else: + raise StorageConfigurationError("key must be bytes or an encoded string") + + if len(raw) != KEY_SIZE: + raise StorageConfigurationError("storage key must decode to exactly 32 bytes") + return raw + + +def _decode_base64(text: str) -> bytes: + if not text: + raise StorageConfigurationError("storage key is empty") + try: + encoded = text.encode("ascii", errors="strict") + except UnicodeEncodeError as exc: + raise StorageConfigurationError("storage key is not valid base64") from exc + encoded += b"=" * (-len(encoded) % 4) + try: + return base64.b64decode(encoded, altchars=b"-_", validate=True) + except (ValueError, UnicodeEncodeError, base64.binascii.Error) as exc: + raise StorageConfigurationError("storage key is not valid base64") from exc + + +def _parse_size(value: int | str) -> int: + if isinstance(value, bool): + raise StorageConfigurationError("max_file_size must be a positive integer or size string") + if isinstance(value, int): + result = value + elif isinstance(value, str): + text = value.strip().replace("_", "").upper() + match = fullmatch( + r"([0-9]+(?:\.[0-9]+)?)\s*(B|KB|KIB|MB|MIB|GB|GIB|TB|TIB)?", text, flags=0 + ) + if not match: + raise StorageConfigurationError( + "invalid size; use bytes or values such as 64MiB, 1GB, or 500KB" + ) + number, suffix = match.groups() + multipliers = { + None: 1, + "B": 1, + "KB": 1_000, + "MB": 1_000_000, + "GB": 1_000_000_000, + "TB": 1_000_000_000_000, + "KIB": 1 << 10, + "MIB": 1 << 20, + "GIB": 1 << 30, + "TIB": 1 << 40, + } + result = int(float(number) * multipliers[suffix]) + else: + raise StorageConfigurationError("max_file_size must be an integer or size string") + if result <= 0: + raise StorageConfigurationError("max_file_size must be greater than zero") + return result + + +def _parse_nonnegative_int(value: int | str, name: str) -> int: + if isinstance(value, bool): + raise StorageConfigurationError(f"{name} must be a non-negative integer") + try: + result = int(value) + except (TypeError, ValueError) as exc: + raise StorageConfigurationError(f"{name} must be a non-negative integer") from exc + if result < 0: + raise StorageConfigurationError(f"{name} must be a non-negative integer") + return result + + +def _parse_timeout(value: float | int | str) -> float: + try: + result = float(value) + except (TypeError, ValueError) as exc: + raise StorageConfigurationError("lock_timeout must be a non-negative number") from exc + if result < 0: + raise StorageConfigurationError("lock_timeout must be a non-negative number") + return result + + +def _parse_bool(value: bool | str) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + text = value.strip().lower() + if text in {"1", "true", "yes", "on", "y"}: + return True + if text in {"0", "false", "no", "off", "n"}: + return False + raise StorageConfigurationError("compression must be true/false, yes/no, or 1/0") + + +def _derive_keys(master_key: bytes) -> tuple[bytes, bytes]: + derived = HKDF( + algorithm=hashes.SHA256(), + length=KEY_SIZE * 2, + salt=None, + info=b"ez-storage/v2 key separation", + ).derive(master_key) + return derived[:KEY_SIZE], derived[KEY_SIZE:] + + +def _validate_pickle_opcodes(payload: bytes) -> None: + """Reject unsupported Pickle protocols, opcodes and trailing bytes.""" + + seen_protocol = False + stop_position: int | None = None + try: + for opcode, arg, position in pickletools.genops(payload): + name = opcode.name + if name not in _ALLOWED_OPCODES: + raise UnsafeTypeError(f"Pickle opcode {name!r} is not allowed") + if name == "PROTO": + if seen_protocol or arg not in {4, 5}: + raise StorageFormatError("only Pickle protocol 4 or 5 is supported") + seen_protocol = True + elif name == "STOP": + stop_position = position + if not seen_protocol: + raise StorageFormatError("Pickle payload has no supported protocol marker") + if stop_position is None or stop_position + 1 != len(payload): + raise StorageFormatError("Pickle payload has no valid final STOP opcode") + except (StorageFormatError, UnsafeTypeError): + raise + except Exception as exc: + raise StorageFormatError("Pickle opcode validation failed") from exc + + +class _RestrictedUnpickler(pickle.Unpickler): + def __init__(self, file: io.BytesIO, allowed_globals: Mapping[tuple[str, str], Any]): + super().__init__(file) + self._allowed_globals = allowed_globals + + def find_class(self, module: str, name: str) -> Any: + key = (module, name) + try: + return self._allowed_globals[key] + except KeyError as exc: + raise UnsafeTypeError( + f"Pickle global {module}.{name} is not registered; " + "register the class explicitly before loading" + ) from exc + + +class _FileLock: + """Small standard-library cross-platform advisory file lock.""" + + def __init__(self, path: Path, timeout: float): + self.path = path + self.timeout = timeout + self._handle: Any = None + + def acquire(self) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + handle = self.path.open("a+b") + if os.name == "nt": + handle.seek(0, os.SEEK_END) + if handle.tell() == 0: + handle.write(b"\0") + handle.flush() + handle.seek(0) + + deadline = time.monotonic() + self.timeout + while True: + try: + if os.name == "nt": + import msvcrt + + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + self._handle = handle + return + except (BlockingIOError, OSError) as exc: + if time.monotonic() >= deadline: + handle.close() + raise StorageLockError( + f"could not lock {self.path} within {self.timeout:g} seconds" + ) from exc + time.sleep(0.05) + + def release(self) -> None: + handle, self._handle = self._handle, None + if handle is None: + return + try: + if os.name == "nt": + import msvcrt + + handle.seek(0) + msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + finally: + handle.close() + + def __enter__(self) -> _FileLock: + self.acquire() + return self + + def __exit__(self, *_: Any) -> None: + self.release() + + +class SecureStore: + """A simple encrypted object store backed by one authenticated file. + + The store reads configuration using this precedence: + + ``per-call override > constructor value > environment > default``. + + The encryption key has no default. Files are only decoded with safe + built-in values and classes explicitly registered with ``register_type``. + """ + + _ENV_NAMES: ClassVar[dict[str, str]] = { + "path": "PATH", + "key": "KEY", + "max_file_size": "MAX_FILE_SIZE", + "backups": "BACKUPS", + "lock_timeout": "LOCK_TIMEOUT", + "compression": "COMPRESSION", + } + + def __init__( + self, + path: str | os.PathLike[str] | None = None, + *, + key: bytes | bytearray | memoryview | str | None = None, + max_file_size: int | str | None = None, + backups: int | str | None = None, + lock_timeout: float | int | str | None = None, + compression: bool | str | None = None, + env_prefix: str = "PYLOCKBOX_", + env: Mapping[str, str] | None = None, + allowed_types: Iterable[type[Any]] | None = None, + ): + if not isinstance(env_prefix, str): + raise StorageConfigurationError("env_prefix must be a string") + self._path = Path(path) if path is not None else None + self._key = key + self._max_file_size = max_file_size + self._backups = backups + self._lock_timeout = lock_timeout + self._compression = compression + self._env_prefix = env_prefix + self._env: Mapping[str, str] = os.environ if env is None else env + self._allowed_globals: dict[tuple[str, str], Any] = dict(_SAFE_BUILTINS) + self._data: Any = _MISSING + + if allowed_types is not None: + for type_ in allowed_types: + self.register_type(type_) + + @classmethod + def from_env( + cls, + *, + prefix: str = "PYLOCKBOX_", + env: Mapping[str, str] | None = None, + **overrides: Any, + ) -> SecureStore: + """Create a store whose unset options are read from ``env``.""" + + return cls(env_prefix=prefix, env=env, **overrides) + + @staticmethod + def generate_key() -> bytes: + """Return a new random 256-bit key.""" + + return generate_key() + + @staticmethod + def encode_key(key: bytes | bytearray | memoryview) -> str: + """Return a URL-safe base64 key suitable for an environment variable.""" + + return encode_key(key) + + @property + def path(self) -> Path: + """The currently resolved storage path.""" + + return self._resolve_path() + + @property + def data(self) -> Any: + """The in-memory value, or ``None`` until a value has been loaded.""" + + return None if self._data is _MISSING else self._data + + def register_type(self, type_: type[_T]) -> type[_T]: + """Allow one user-defined class to be decoded by this store. + + Registration is intentionally explicit and local to this instance; + the loader never imports a module based on data found in the file. + """ + + if not isinstance(type_, type): + raise StorageConfigurationError("register_type expects a class") + module = getattr(type_, "__module__", "") + name = getattr(type_, "__qualname__", "") + if not module or not name or "" in name: + raise StorageConfigurationError( + "registered classes must be top-level classes with a stable qualified name" + ) + self._allowed_globals[(module, name)] = type_ + return type_ + + def configure( + self, + *, + path: str | os.PathLike[str] | None | object = _MISSING, + key: bytes | bytearray | memoryview | str | None | object = _MISSING, + max_file_size: int | str | None | object = _MISSING, + backups: int | str | None | object = _MISSING, + lock_timeout: float | int | str | None | object = _MISSING, + compression: bool | str | None | object = _MISSING, + ) -> SecureStore: + """Update constructor-level settings and return ``self``.""" + + if path is not _MISSING: + self._path = None if path is None else Path(path) # type: ignore[arg-type] + if key is not _MISSING: + self._key = key # type: ignore[assignment] + if max_file_size is not _MISSING: + self._max_file_size = max_file_size # type: ignore[assignment] + if backups is not _MISSING: + self._backups = backups # type: ignore[assignment] + if lock_timeout is not _MISSING: + self._lock_timeout = lock_timeout # type: ignore[assignment] + if compression is not _MISSING: + self._compression = compression # type: ignore[assignment] + return self + + def exists(self, *, path: str | os.PathLike[str] | None = None) -> bool: + """Return whether the resolved storage file exists.""" + + return self._resolve_path(path).is_file() + + def load( + self, + default: Any = _MISSING, + *, + path: str | os.PathLike[str] | None = None, + key: bytes | bytearray | memoryview | str | None = None, + max_file_size: int | str | None = None, + lock_timeout: float | int | str | None = None, + ) -> Any: + """Load and return the root object. + + ``key``, ``max_file_size`` and the other constructor settings can be + overridden for this call. If the file does not yet exist, ``default`` + is returned; without an explicit default this is an empty dictionary. + """ + + resolved_path = self._resolve_path(path) + resolved_key = self._resolve_key(key) + size_limit = self._resolve_max_file_size(max_file_size) + timeout = self._resolve_lock_timeout(lock_timeout) + + with self._locked(resolved_path, timeout): + if not resolved_path.exists(): + value = {} if default is _MISSING else default + else: + encoded = self._read_limited(resolved_path, size_limit) + value = self._decode(encoded, resolved_key, size_limit) + + self._data = value + return value + + def reload(self, *args: Any, **kwargs: Any) -> Any: + """Alias for :meth:`load`.""" + + return self.load(*args, **kwargs) + + def save( + self, + value: Any = _MISSING, + *, + path: str | os.PathLike[str] | None = None, + key: bytes | bytearray | memoryview | str | None = None, + max_file_size: int | str | None = None, + backups: int | str | None = None, + lock_timeout: float | int | str | None = None, + compression: bool | str | None = None, + ) -> None: + """Encrypt and atomically save ``value``. + + If ``value`` is omitted, the current in-memory object is saved. A + store that has not been loaded starts with an empty dictionary. + """ + + resolved_path = self._resolve_path(path) + resolved_key = self._resolve_key(key) + size_limit = self._resolve_max_file_size(max_file_size) + backup_count = self._resolve_backups(backups) + timeout = self._resolve_lock_timeout(lock_timeout) + use_compression = self._resolve_compression(compression) + + if value is _MISSING: + if self._data is _MISSING: + value = self.load( + path=resolved_path, + key=resolved_key, + max_file_size=size_limit, + lock_timeout=timeout, + ) + else: + value = self._data + + encoded = self._encode(value, resolved_key, size_limit, use_compression) + with self._locked(resolved_path, timeout): + if resolved_path.exists() and backup_count: + self._rotate_backups(resolved_path, backup_count) + self._atomic_write(resolved_path, encoded) + self._data = value + + def get(self, name: str, default: Any = None) -> Any: + """Get a top-level or dotted mapping key from the loaded object.""" + + mapping = self._ensure_mapping() + try: + return self._lookup(mapping, name) + except KeyError: + return default + + def set(self, name: str, value: Any, *, autosave: bool = False, **save_options: Any) -> SecureStore: + """Set a top-level or dotted mapping key.""" + + mapping = self._ensure_mapping() + parts = self._split_key(name) + current: dict[str, Any] = mapping + for part in parts[:-1]: + if part not in current: + existing = {} + current[part] = existing + else: + existing = current[part] + if not isinstance(existing, dict): + raise StorageTypeError(f"cannot descend into non-mapping key {part!r}") + current = existing + current[parts[-1]] = value + if autosave: + self.save(**save_options) + return self + + def remove(self, name: str, *, autosave: bool = False, **save_options: Any) -> bool: + """Remove a mapping key and return whether it existed.""" + + mapping = self._ensure_mapping() + parts = self._split_key(name) + current: Any = mapping + for part in parts[:-1]: + if not isinstance(current, dict) or part not in current: + return False + current = current[part] + if not isinstance(current, dict) or parts[-1] not in current: + return False + del current[parts[-1]] + if autosave: + self.save(**save_options) + return True + + def clear(self, *, autosave: bool = False, **save_options: Any) -> SecureStore: + """Clear the in-memory mapping.""" + + self._data = {} + if autosave: + self.save(**save_options) + return self + + def keys(self) -> tuple[Any, ...]: + """Return root mapping keys.""" + + return tuple(self._ensure_mapping().keys()) + + def items(self) -> tuple[tuple[Any, Any], ...]: + """Return root mapping items.""" + + return tuple(self._ensure_mapping().items()) + + def list_backups(self, *, path: str | os.PathLike[str] | None = None) -> tuple[Path, ...]: + """Return existing backup paths, newest first.""" + + resolved_path = self._resolve_path(path) + backups = list(resolved_path.parent.glob(f"{resolved_path.name}.bak.*")) + return tuple( + sorted( + (item for item in backups if item.name.rsplit(".bak.", 1)[-1].isdigit()), + key=lambda item: int(item.name.rsplit(".bak.", 1)[-1]), + ) + ) + + def delete(self, *, path: str | os.PathLike[str] | None = None) -> bool: + """Delete only the main storage file, leaving backups untouched.""" + + resolved_path = self._resolve_path(path) + timeout = self._resolve_lock_timeout(None) + with self._locked(resolved_path, timeout): + try: + resolved_path.unlink() + except FileNotFoundError: + return False + self._data = _MISSING + return True + + def _env_value(self, setting: str) -> str | None: + return self._env.get(f"{self._env_prefix}{self._ENV_NAMES[setting]}") + + def _resolve_path(self, override: str | os.PathLike[str] | None | object = None) -> Path: + if override is not None: + return Path(override) # type: ignore[arg-type] + if self._path is not None: + return self._path + env_value = self._env_value("path") + return Path(env_value) if env_value else Path(DEFAULT_PATH) + + def _resolve_key( + self, override: bytes | bytearray | memoryview | str | None = None + ) -> bytes: + value: bytes | bytearray | memoryview | str | None = override + if value is None: + value = self._key + if value is None: + value = self._env_value("key") + if value is None: + raise MissingKeyError( + f"no encryption key configured; set {self._env_prefix}{self._ENV_NAMES['key']} " + "or pass key=..." + ) + return _normalise_key(value) + + def _resolve_max_file_size(self, override: int | str | None) -> int: + value: int | str | None = override + if value is None: + value = self._max_file_size + if value is None: + value = self._env_value("max_file_size") + return _parse_size(DEFAULT_MAX_FILE_SIZE if value is None else value) + + def _resolve_backups(self, override: int | str | None) -> int: + value: int | str | None = override + if value is None: + value = self._backups + if value is None: + value = self._env_value("backups") + return _parse_nonnegative_int(DEFAULT_BACKUPS if value is None else value, "backups") + + def _resolve_lock_timeout(self, override: float | int | str | None) -> float: + value: float | int | str | None = override + if value is None: + value = self._lock_timeout + if value is None: + value = self._env_value("lock_timeout") + return _parse_timeout(DEFAULT_LOCK_TIMEOUT if value is None else value) + + def _resolve_compression(self, override: bool | str | None) -> bool: + value: bool | str | None = override + if value is None: + value = self._compression + if value is None: + value = self._env_value("compression") + return _parse_bool(DEFAULT_COMPRESSION if value is None else value) + + @contextmanager + def _locked(self, path: Path, timeout: float) -> Iterator[None]: + lock_path = Path(f"{path}.lock") + with _FileLock(lock_path, timeout): + yield + + @staticmethod + def _read_limited(path: Path, size_limit: int) -> bytes: + try: + if path.stat().st_size > size_limit: + raise StorageSizeError( + f"storage file is larger than the configured limit of {size_limit} bytes" + ) + with path.open("rb") as handle: + data = handle.read(size_limit + 1) + except FileNotFoundError: + raise + if len(data) > size_limit: + raise StorageSizeError( + f"storage file grew beyond the configured limit of {size_limit} bytes" + ) + return data + + @staticmethod + def _encode(value: Any, key: bytes, size_limit: int, compression: bool) -> bytes: + try: + payload = pickle.dumps(value, protocol=pickle.HIGHEST_PROTOCOL) + except (AttributeError, OverflowError, pickle.PickleError, TypeError, ValueError) as exc: + raise StorageConfigurationError("value cannot be serialized with Pickle") from exc + _validate_pickle_opcodes(payload) + + flags = 0 + if compression: + compressed = zlib.compress(payload, level=9) + if len(compressed) < len(payload): + payload = compressed + flags |= _FLAG_COMPRESSED + + if len(payload) + _ENVELOPE_OVERHEAD > size_limit: + raise StorageSizeError( + f"serialized value is larger than the configured limit of {size_limit} bytes" + ) + + nonce = secrets.token_bytes(NONCE_SIZE) + ciphertext_size = len(payload) + 16 + header = _HEADER.pack( + _MAGIC, + _FORMAT_VERSION, + flags, + _HEADER.size, + time.time_ns() & ((1 << 64) - 1), + ciphertext_size, + nonce, + ) + encryption_key, mac_key = _derive_keys(key) + ciphertext = ChaCha20Poly1305(encryption_key).encrypt(nonce, payload, header) + tag = hmac.new(mac_key, header + ciphertext, hashlib.sha256).digest() + return header + tag + ciphertext + + def _decode(self, encoded: bytes, key: bytes, size_limit: int) -> Any: + if len(encoded) > size_limit: + raise StorageSizeError("storage file exceeds the configured limit") + if len(encoded) < _HEADER.size + HMAC_SIZE + 16: + raise StorageFormatError("storage file is truncated") + + header = encoded[: _HEADER.size] + try: + magic, version, flags, header_size, _generation, payload_size, nonce = _HEADER.unpack(header) + except struct.error as exc: + raise StorageFormatError("invalid storage header") from exc + if magic != _MAGIC: + raise StorageFormatError("invalid PyLockbox magic header") + if version != _FORMAT_VERSION: + raise StorageFormatError(f"unsupported PyLockbox format version: {version}") + if header_size != _HEADER.size: + raise StorageFormatError("invalid PyLockbox header size") + if flags & ~_KNOWN_FLAGS: + raise StorageFormatError("storage file uses unknown flags") + if payload_size < 16: + raise StorageFormatError("invalid encrypted payload size") + expected_size = _HEADER.size + HMAC_SIZE + payload_size + if expected_size != len(encoded): + raise StorageFormatError("storage file has an invalid payload length") + + tag_start = _HEADER.size + tag_end = tag_start + HMAC_SIZE + tag = encoded[tag_start:tag_end] + ciphertext = encoded[tag_end:] + encryption_key, mac_key = _derive_keys(key) + expected_tag = hmac.new(mac_key, header + ciphertext, hashlib.sha256).digest() + if not hmac.compare_digest(tag, expected_tag): + raise StorageIntegrityError("storage authentication failed (wrong key or modified file)") + + try: + payload = ChaCha20Poly1305(encryption_key).decrypt(nonce, ciphertext, header) + except InvalidTag as exc: + raise StorageIntegrityError("storage encryption authentication failed") from exc + + if len(payload) > size_limit: + raise StorageSizeError("decoded payload exceeds the configured limit") + if flags & _FLAG_COMPRESSED: + decompressor = zlib.decompressobj() + payload = decompressor.decompress(payload, size_limit + 1) + if len(payload) > size_limit or decompressor.unconsumed_tail: + raise StorageSizeError("decompressed payload exceeds the configured limit") + payload += decompressor.flush() + if len(payload) > size_limit: + raise StorageSizeError("decompressed payload exceeds the configured limit") + if not decompressor.eof: + raise StorageFormatError("compressed payload is incomplete") + + _validate_pickle_opcodes(payload) + try: + return _RestrictedUnpickler(io.BytesIO(payload), self._allowed_globals).load() + except (StorageFormatError, UnsafeTypeError): + raise + except Exception as exc: + raise StorageFormatError("restricted Pickle loading failed") from exc + + @staticmethod + def _atomic_write(path: Path, content: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{path.name}.", suffix=".tmp", dir=path.parent + ) + temporary_path = Path(temporary_name) + try: + try: + os.chmod(temporary_path, 0o600) + except OSError: + pass + with os.fdopen(descriptor, "wb") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, path) + try: + os.chmod(path, 0o600) + except OSError: + pass + try: + directory_fd = os.open(path.parent, os.O_RDONLY) + except (OSError, TypeError): + directory_fd = None + if directory_fd is not None: + try: + os.fsync(directory_fd) + except OSError: + pass + finally: + os.close(directory_fd) + finally: + try: + temporary_path.unlink() + except FileNotFoundError: + pass + + @staticmethod + def _rotate_backups(path: Path, count: int) -> None: + for index in range(count, 1, -1): + source = Path(f"{path}.bak.{index - 1}") + destination = Path(f"{path}.bak.{index}") + if source.exists(): + os.replace(source, destination) + backup = Path(f"{path}.bak.1") + shutil.copy2(path, backup) + try: + os.chmod(backup, 0o600) + except OSError: + pass + + def _ensure_mapping(self) -> dict[str, Any]: + if self._data is _MISSING: + self.load(default={}) + if not isinstance(self._data, dict): + raise StorageTypeError("key operations require the stored root object to be a dict") + return self._data + + @staticmethod + def _split_key(name: str) -> list[str]: + if not isinstance(name, str) or not name or name.startswith(".") or name.endswith("."): + raise StorageConfigurationError("mapping key must be a non-empty string") + parts = name.split(".") + if any(not part for part in parts): + raise StorageConfigurationError("mapping key contains an empty path segment") + return parts + + @classmethod + def _lookup(cls, mapping: Mapping[str, Any], name: str) -> Any: + current: Any = mapping + for part in cls._split_key(name): + if not isinstance(current, Mapping): + raise KeyError(name) + current = current[part] + return current + + +__all__ = ["SecureStore", "encode_key", "generate_key"] diff --git a/tests/TEST.py b/tests/TEST.py deleted file mode 100644 index ff746a1..0000000 --- a/tests/TEST.py +++ /dev/null @@ -1,45 +0,0 @@ -from src.ez_storage.ez_storage import Ez_Storage -import logging - -# logger -logger = logging.getLogger() -logging.basicConfig(filename='test.log', encoding='utf-8', level=logging.DEBUG, filemode="w") - -test_add = True -test_get = True -test_restore = False -default = Ez_Storage() -if test_add: - logger.debug("starting: test_add") - new_array = {"source": 1, "destination": 2} - new_list = [1, 2] - override = False - default.add_storage(mode="o", obj="TEST_o", data="test_daten", value="value", override=override) - default.add_storage(mode="a", obj="TEST_a", array_data=new_array, override=override) - default.add_storage(mode="l", obj="TEST_l", data=new_list, override=override) - -if test_get: - logger.debug("starting: test_get") - logger.debug("result 'o' :" + str(default.get_storage(mode="o", obj="TEST_o", data="test_daten"))) - logger.debug("result 'a' :" + str(default.get_storage(mode="a", obj="TEST_a"))) - logger.debug("result 'l' :" + str(default.get_storage(mode="l", obj="TEST_l"))) - -if test_restore: - logger.debug("starting: test_restore") - # restore with base settings - default_b = Ez_Storage("b") - default_b.restore_storage(source="b") - # restore with internal_config in object mode - default_io = Ez_Storage("io") - default_io.internal_config = [("test", "io"), ("object", "object")] - default_io.restore_storage(source="i", destination="Test_IO", mode="o") - # restore with internal_config in array mode - default_ia = Ez_Storage("ia") - new_array = {"source": 1, "destination": 2} - default_ia.internal_config = new_array - default_ia.restore_storage(source="i", destination="Test_IA", mode="a") - # restore with internal_config in list mode - default_il = Ez_Storage("il") - default_il.internal_config = [1, 2] - default_il.restore_storage(source="i", destination="Test_IL", mode="l") - diff --git a/tests/test_secure_store.py b/tests/test_secure_store.py new file mode 100644 index 0000000..d110ea1 --- /dev/null +++ b/tests/test_secure_store.py @@ -0,0 +1,160 @@ +from __future__ import annotations + +import os +import tempfile +import unittest +from dataclasses import dataclass +from pathlib import Path + +from pylockbox import ( + MissingKeyError, + SecureStore, + StorageFormatError, + StorageIntegrityError, + StorageSizeError, + UnsafeTypeError, + encode_key, + generate_key, +) + + +@dataclass +class ExampleState: + name: str + count: int + + +class DangerousValue: + def __init__(self, marker: str): + self.marker = marker + + def __reduce__(self): + return (os.system, (f"touch {self.marker}",)) + + +class SecureStoreTests(unittest.TestCase): + def test_encrypted_round_trip_and_mapping_api(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "state.lockbox" + key = generate_key() + + store = SecureStore(path, key=key, backups=2) + store.set("profile.name", "Lainup").set("counter", 1).save() + + encoded = path.read_bytes() + self.assertEqual(encoded[:4], b"EZST") + self.assertNotIn(b"Lainup", encoded) + + loaded = SecureStore(path, key=encode_key(key)).load() + self.assertEqual(loaded, {"profile": {"name": "Lainup"}, "counter": 1}) + + reloaded_store = SecureStore(path, key=key) + self.assertEqual(reloaded_store.get("profile.name"), "Lainup") + self.assertEqual(reloaded_store.get("missing", "fallback"), "fallback") + + def test_wrong_key_and_tampering_are_rejected(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "state.lockbox" + key = generate_key() + SecureStore(path, key=key).save({"secret": "value"}) + original = path.read_bytes() + + with self.assertRaises(StorageIntegrityError): + SecureStore(path, key=generate_key()).load() + + tampered = bytearray(original) + tampered[-1] ^= 0x01 + path.write_bytes(tampered) + with self.assertRaises(StorageIntegrityError): + SecureStore(path, key=key).load() + + def test_backups_rotate(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "state.lockbox" + store = SecureStore(path, key=generate_key(), backups=2) + store.save({"version": 1}) + store.save({"version": 2}) + store.save({"version": 3}) + + backups = store.list_backups() + self.assertEqual([item.name for item in backups], ["state.lockbox.bak.1", "state.lockbox.bak.2"]) + + def test_restricted_loader_requires_explicit_class_registration(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "state.lockbox" + key = generate_key() + SecureStore(path, key=key).save({"state": ExampleState("demo", 3)}) + + with self.assertRaises(UnsafeTypeError): + SecureStore(path, key=key).load() + + trusted = SecureStore(path, key=key) + trusted.register_type(ExampleState) + self.assertEqual(trusted.load()["state"], ExampleState("demo", 3)) + + def test_restricted_loader_does_not_execute_arbitrary_reduce(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + marker = str(Path(temporary_directory) / "should-not-exist") + path = Path(temporary_directory) / "state.lockbox" + key = generate_key() + SecureStore(path, key=key).save(DangerousValue(marker)) + + with self.assertRaises(UnsafeTypeError): + SecureStore(path, key=key).load() + self.assertFalse(Path(marker).exists()) + + def test_environment_configuration_and_load_overrides(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "from-env.lockbox" + key = generate_key() + environment = { + "APP_STORAGE_PATH": str(path), + "APP_STORAGE_KEY": encode_key(key), + "APP_STORAGE_MAX_FILE_SIZE": "1MiB", + "APP_STORAGE_BACKUPS": "0", + "APP_STORAGE_COMPRESSION": "off", + } + + store = SecureStore.from_env(prefix="APP_STORAGE_", env=environment) + store.save({"from": "environment"}) + loaded = SecureStore.from_env(prefix="APP_STORAGE_", env=environment).load() + self.assertEqual(loaded, {"from": "environment"}) + + other_key = generate_key() + with self.assertRaises(StorageIntegrityError): + store.load(key=other_key) + self.assertEqual(store.load(key=key, max_file_size="2MiB"), {"from": "environment"}) + + def test_pylockbox_is_the_default_environment_prefix(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "default-prefix.lockbox" + key = generate_key() + environment = { + "PYLOCKBOX_PATH": str(path), + "PYLOCKBOX_KEY": encode_key(key), + } + + store = SecureStore.from_env(env=environment) + store.save({"prefix": "pylockbox"}) + self.assertEqual(SecureStore.from_env(env=environment).load(), {"prefix": "pylockbox"}) + + def test_size_limit_and_missing_key(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "state.lockbox" + with self.assertRaises(MissingKeyError): + SecureStore(path).save({"secret": "value"}) + + random_payload = os.urandom(4096) + with self.assertRaises(StorageSizeError): + SecureStore(path, key=generate_key(), max_file_size="1KiB").save(random_payload) + + def test_invalid_format_is_rejected(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + path = Path(temporary_directory) / "state.lockbox" + path.write_bytes(b"not-a-pylockbox-file") + with self.assertRaises(StorageFormatError): + SecureStore(path, key=generate_key()).load() + + +if __name__ == "__main__": + unittest.main() From ca45ab11651f35e3edd7dbe8ac53a7aca4e12cdf Mon Sep 17 00:00:00 2001 From: Sandro <34354689+Lainupcomputer@users.noreply.github.com> Date: Fri, 18 Sep 2026 05:03:46 +0200 Subject: [PATCH 2/2] Update README.md --- README.md | 198 +++++++++++++++++++++++++++++++++++++++--------------- 1 file changed, 145 insertions(+), 53 deletions(-) diff --git a/README.md b/README.md index 956683f..b6a9754 100644 --- a/README.md +++ b/README.md @@ -1,69 +1,116 @@ +
+ # PyLockbox -`PyLockbox` is a small general-purpose object store for Python applications. It stores one -Pickle object in an encrypted, authenticated binary file and provides a simple key/value API for -the common dictionary case. +**Secure, encrypted and easy-to-use object storage for Python applications.** + +Encrypted Pickle storage · authenticated files · restricted loading · environment configuration + +[![CI](https://github.com/Lainupcomputer/PyLockbox/actions/workflows/ci.yml/badge.svg?branch=v2)](https://github.com/Lainupcomputer/PyLockbox/actions/workflows/ci.yml) +![Python](https://img.shields.io/badge/Python-3.10%2B-3776AB?logo=python&logoColor=white) +![Version](https://img.shields.io/badge/Version-2.0.0-7C3AED) +![Security](https://img.shields.io/badge/Security-ChaCha20--Poly1305-8B5CF6) +![License](https://img.shields.io/badge/License-MIT-22C55E) + +[Getting Started](#getting-started) · [Environment Variables](#environment-variables) · [Security Model](#security-model) · [API](#api-overview) · [Development](#development) + +
+ +--- + +PyLockbox is a general-purpose storage library for Python. It stores application state in a single encrypted and authenticated binary file while providing a simple dictionary-style API for common key/value data. + +The package is designed for configuration, sessions, local application state, caches, and other data that should be protected at rest without requiring a database. + +> [!IMPORTANT] +> Version 2.0.0 is a breaking rewrite. PyLockbox v2 intentionally does not read the previous `ez-storage` JSON format and does not include a legacy migration layer. + +## Features -Version 2 is a breaking rewrite. It intentionally does not read the old JSON format and does not -include a legacy migration layer. +- Encrypted binary storage using **ChaCha20-Poly1305**. +- Separate encryption and integrity keys derived with **HKDF**. +- Additional **HMAC-SHA256** authentication for the envelope and ciphertext. +- Restricted Pickle loading with safe built-ins and explicit class registration. +- No dynamic imports based on data stored in a file. +- Atomic writes with temporary files, `fsync`, and replacement. +- Cross-platform file locking for concurrent access. +- Rotating backups with configurable retention. +- File-size and decompression limits to reduce resource-exhaustion risks. +- Optional compression before encryption. +- Environment-variable configuration with custom prefixes. +- Per-call configuration overrides without changing the store defaults. +- Simple dotted-key mapping access such as `profile.name`. +- Python 3.10–3.13 support. -## Install +## Getting Started + +### Requirements + +- Python **3.10 or newer** +- `cryptography >= 42` + +### Installation + +Install the released package from PyPI: ```bash python -m pip install pylockbox ``` -For development: +For local development: ```bash +git clone https://github.com/Lainupcomputer/PyLockbox.git +cd PyLockbox python -m pip install -e ".[dev]" ``` -## Quick start +### Quick start -The key must be supplied as raw 32-byte data or as base64/hex text. There is no default key. +PyLockbox requires a 32-byte encryption key. There is no insecure default key. ```python from pylockbox import SecureStore, encode_key, generate_key key = generate_key() -print("Store this outside your repository:", encode_key(key)) +print("Store this key outside your repository:", encode_key(key)) store = SecureStore("state.lockbox", key=key) -store.set("user.name", "Lainup") +store.set("profile.name", "Lainup") store.set("settings.theme", "dark") +store.set("launch_count", 1) store.save() reader = SecureStore("state.lockbox", key=key) print(reader.load()) -print(reader.get("user.name")) +print(reader.get("profile.name")) ``` -`save(value)` and `load()` also work with any Pickle-compatible root value. The mapping helpers -(`set`, `get`, `remove`, `keys`, and `items`) require the root value to be a dictionary. Dotted -names such as `settings.theme` address nested dictionaries. +The generated key must be stored securely. Do not commit it to GitHub or place it directly in source code used in production. + +`save(value)` and `load()` also support Pickle-compatible root objects. The dotted-key helpers require the root object to be a dictionary. -## Environment variables +## Environment Variables -Use `SecureStore.from_env()` to load configuration from environment variables. The default prefix -is `PYLOCKBOX_`; a custom prefix is supported for applications with multiple stores. +Use `SecureStore.from_env()` when configuration should come from the process environment. The default prefix is `PYLOCKBOX_`. -| Variable | Default | Meaning | +| Variable | Default | Description | | --- | --- | --- | | `PYLOCKBOX_PATH` | `storage.lockbox` | Storage file path | | `PYLOCKBOX_KEY` | required | URL-safe base64 or hex encoded 32-byte key | -| `PYLOCKBOX_MAX_FILE_SIZE` | `64MiB` | Maximum file and decoded payload size | +| `PYLOCKBOX_MAX_FILE_SIZE` | `64MiB` | Maximum storage and decoded payload size | | `PYLOCKBOX_BACKUPS` | `3` | Number of rotating backups; `0` disables backups | | `PYLOCKBOX_LOCK_TIMEOUT` | `5` | Lock wait time in seconds | -| `PYLOCKBOX_COMPRESSION` | `true` | Compress before encryption when it saves space | +| `PYLOCKBOX_COMPRESSION` | `true` | Compress when compression reduces the payload size | -Example in PowerShell: +Example in Windows PowerShell: ```powershell $env:PYLOCKBOX_PATH = "state.lockbox" $env:PYLOCKBOX_KEY = "paste-a-generated-base64-key-here" $env:PYLOCKBOX_MAX_FILE_SIZE = "128MiB" $env:PYLOCKBOX_BACKUPS = "5" +$env:PYLOCKBOX_COMPRESSION = "true" ``` ```python @@ -71,6 +118,13 @@ from pylockbox import SecureStore store = SecureStore.from_env() store.set("counter", 1).save() +value = store.load() +``` + +Custom prefixes are useful when an application manages more than one store: + +```python +store = SecureStore.from_env(prefix="APP_STORAGE_") ``` Configuration precedence is: @@ -79,50 +133,50 @@ Configuration precedence is: per-call override > constructor argument > environment variable > default ``` -For example, a one-time load override does not change the store configuration: +For example, a one-time load override does not modify the store configuration: ```python value = store.load(key=another_key, max_file_size="256MiB") ``` -The library reads environment variables; it never writes secrets back into `os.environ`. +PyLockbox reads environment variables but never writes secrets back into `os.environ`. -## Security model +## Security Model -Each file contains a versioned binary envelope, not JSON: +Every storage file uses a versioned binary envelope rather than JSON: -- ChaCha20-Poly1305 encrypts and authenticates the payload. -- HKDF separates the encryption and HMAC keys derived from the supplied 32-byte key. -- HMAC-SHA256 authenticates the header and ciphertext as an additional integrity layer. -- Pickle protocol 4/5 opcodes are checked before loading. -- A restricted unpickler allows safe built-in values and classes explicitly registered in code. -- No module is imported because a file requests it; unknown globals and persistent IDs are rejected. -- Maximum file size is checked before reading, and decompression is bounded. -- Writes use a temporary file, `fsync`, and atomic replacement. A cross-platform file lock protects - concurrent readers/writers, and POSIX files are created with mode `0600` where supported. +1. Pickle serializes the application value. +2. Supported Pickle opcodes are validated before loading. +3. The payload is optionally compressed. +4. ChaCha20-Poly1305 encrypts and authenticates the payload. +5. HMAC-SHA256 authenticates the header and ciphertext as an additional integrity layer. +6. A restricted unpickler loads only safe built-ins or classes explicitly registered by the application. + +The implementation also applies size limits before reading, bounds decompression, uses atomic file replacement, and protects access with a cross-platform lock. ### Important Pickle limitation -Pickle is not a security sandbox. This package makes loading substantially safer for authenticated -application state, but no implementation can make arbitrary Pickle data completely safe if an -attacker controls the Python process, obtains the key, or is allowed to register a malicious class. -Only load files protected by a key you trust, and register only classes whose deserialization code -you trust. +Pickle is not a security sandbox. PyLockbox is intended for authenticated application state protected by a trusted key; it is not intended for arbitrary files from untrusted people. + +Do not load a file when an attacker controls the Python process, has obtained the encryption key, or can register malicious classes in your application. Register only classes whose deserialization behavior you trust. -## User-defined classes +## User-defined Classes -Custom classes are denied by default. Register the exact class on every process that loads the -file: +Custom classes are denied by default. Register the exact class on every process that loads the file: ```python from dataclasses import dataclass + from pylockbox import SecureStore + @dataclass class Profile: name: str + key = SecureStore.generate_key() + writer = SecureStore("profiles.lockbox", key=key) writer.register_type(Profile) writer.save(Profile("Lainup")) @@ -130,21 +184,43 @@ writer.save(Profile("Lainup")) reader = SecureStore("profiles.lockbox", key=key) reader.register_type(Profile) profile = reader.load() + +print(profile.name) ``` -This explicit registration is deliberate: arbitrary functions, modules, and classes are not loaded -from file data. +Explicit registration prevents file data from freely importing arbitrary modules, classes, or functions. + +## API Overview + +| API | Purpose | +| --- | --- | +| `SecureStore(path, key=...)` | Create a configured store | +| `SecureStore.from_env(...)` | Create a store from environment variables | +| `generate_key()` | Generate a cryptographically random 32-byte key | +| `encode_key(key)` | Encode a key for environment variables | +| `store.save(value)` | Encrypt and save a root value | +| `store.load()` | Verify, decrypt, and safely load a value | +| `store.set("a.b", value)` | Set a nested dictionary value | +| `store.get("a.b", default)` | Read a nested dictionary value | +| `store.remove("a.b")` | Remove a nested dictionary value | +| `store.keys()` / `store.items()` | Inspect dictionary contents | +| `store.register_type(MyClass)` | Explicitly allow a custom class | +| `store.list_backups()` | List rotating backup files | +| `store.delete()` | Delete the main storage file | -## Development and publishing +## Development -Run the tests with the standard library: +Run the test suite with the Python standard library: ```bash python -m unittest discover -s tests -v ``` -The repository includes a build/check/upload script. It does not upload unless `--publish` is -passed, and it never contains a PyPI token: +The GitHub Actions workflow runs the tests on Python 3.10, 3.11, 3.12, and 3.13. + +### Build and validate the package + +The repository includes a publishing helper. It builds and validates the package without uploading anything unless `--publish` is explicitly provided: ```bash # Build and validate only @@ -157,16 +233,32 @@ python scripts/publish.py --repository testpypi --publish python scripts/publish.py --repository pypi --publish ``` -On Windows PowerShell the wrapper can be used instead: +Windows PowerShell wrapper: ```powershell .\scripts\publish.ps1 --clean .\scripts\publish.ps1 --repository testpypi --publish ``` -Configure credentials through Twine's normal mechanisms (`TWINE_USERNAME`, `TWINE_PASSWORD`, a -keyring, or `.pypirc`). Never commit credentials or a generated storage key. +Configure upload credentials through Twine's normal mechanisms such as `TWINE_USERNAME`, `TWINE_PASSWORD`, a keyring, or `.pypirc`. Never commit credentials or generated storage keys. + +## Project Structure + +```text +PyLockbox/ +├── src/pylockbox/ # Package implementation +├── tests/ # Security and behavior tests +├── scripts/ # Build and publishing helpers +├── pyproject.toml # Package metadata and tooling +└── README.md # Documentation +``` ## License -MIT. See [LICENSE](LICENSE). +PyLockbox is released under the [MIT License](LICENSE). + +
+ +Made for Python applications that need simple, protected local storage. + +