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..b6a9754 100644
--- a/README.md
+++ b/README.md
@@ -1,16 +1,264 @@
-# EZ Storage
+
-mostly it is a dependency
+# PyLockbox
-function overview provided in `/tests/TEST.py`
+**Secure, encrypted and easy-to-use object storage for Python applications.**
-## Installation:
+Encrypted Pickle storage · authenticated files · restricted loading · environment configuration
-` python -m pip install ez-storage`
-## Usages:
+[](https://github.com/Lainupcomputer/PyLockbox/actions/workflows/ci.yml)
+
+
+
+
-1. used in [stable-diffusion-webui-random_prompt_generator](https://github.com/Lainupcomputer/stable-diffusion-webui-random_prompt_generator)
+[Getting Started](#getting-started) · [Environment Variables](#environment-variables) · [Security Model](#security-model) · [API](#api-overview) · [Development](#development)
+
-## Notes:
-feel free to use
\ No newline at end of file
+---
+
+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
+
+- 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.
+
+## Getting Started
+
+### Requirements
+
+- Python **3.10 or newer**
+- `cryptography >= 42`
+
+### Installation
+
+Install the released package from PyPI:
+
+```bash
+python -m pip install pylockbox
+```
+
+For local development:
+
+```bash
+git clone https://github.com/Lainupcomputer/PyLockbox.git
+cd PyLockbox
+python -m pip install -e ".[dev]"
+```
+
+### Quick start
+
+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 key outside your repository:", encode_key(key))
+
+store = SecureStore("state.lockbox", key=key)
+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("profile.name"))
+```
+
+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
+
+Use `SecureStore.from_env()` when configuration should come from the process environment. The default prefix is `PYLOCKBOX_`.
+
+| 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 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 when compression reduces the payload size |
+
+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
+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:
+
+```text
+per-call override > constructor argument > environment variable > default
+```
+
+For example, a one-time load override does not modify the store configuration:
+
+```python
+value = store.load(key=another_key, max_file_size="256MiB")
+```
+
+PyLockbox reads environment variables but never writes secrets back into `os.environ`.
+
+## Security Model
+
+Every storage file uses a versioned binary envelope rather than JSON:
+
+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. 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
+
+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()
+
+print(profile.name)
+```
+
+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
+
+Run the test suite with the Python standard library:
+
+```bash
+python -m unittest discover -s tests -v
+```
+
+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
+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
+```
+
+Windows PowerShell wrapper:
+
+```powershell
+.\scripts\publish.ps1 --clean
+.\scripts\publish.ps1 --repository testpypi --publish
+```
+
+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
+
+PyLockbox is released under the [MIT License](LICENSE).
+
+
+
+Made for Python applications that need simple, protected local storage.
+
+
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()