janware/jw-pkg
Watch
1
Code Pull requests Activity Actions

lib.ec.secret_store.Sops: Add store class
All checks were successful
CI / Packaging - Kali Linux (pull_request) Successful in 5m0s
Details
CI / Packaging - OpenSUSE Tumbleweed (pull_request) Successful in 5m16s
Details
CI / Packaging test (pull_request) Successful in 0s
Details
CI / Packaging - Kali Linux (push) Successful in 4m37s
Details
CI / Packaging - OpenSUSE Tumbleweed (push) Successful in 4m57s
Details
CI / Packaging test (push) Successful in 0s
Details

Browse source 
This commit adds a new class inheriting from secret_store.Base: Sops. The
currently available classes allow to store secrets offline in a
GPG-encrypted tarball, but those are not versioned. This commit allows to
use a SOPS / AGE-encrypted Git repository as storage backend instead.

The additions in detail:

- Sops: the store implementation; the repository location comes from
  JW_SECRETS_SOPS_REPO, get() on an entry returns the decrypted serialized
  content, and a typed put (FT_SECRET) re-encrypts the entry wholesale
- The age key comes from JW_SECRETS_SOPS_KEY_PATH only: the house variable
  is the sole input to the store, and SOPS_AGE_KEY_FILE is set on the sops
  subprocess per invocation, never read from the process environment;
  SopsKeyError is raised when the variable is not set
- FileContext.create(): the sops:// scheme
- The store's unit tests, covering the node types, the deploy and fetch
  round trip, the whole-entry get and put, the key resolution, and the
  default fallback

Signed-off-by: Jan Lindemann <jan@janware.com>
Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.1.0
Signed-off-by: Jan Lindemann <jan@janware.com>
Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.1.0
This commit is contained in:
Jan Lindemann 2026-10-03 21:39:33 +02:00
commit 60a91c0692
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
2 changed files with 1066 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/python/jw/pkg/lib/FileContext.py
View file

@ -360,6 +360,10 @@ class FileContext(abc.ABC):
from .ec.Curl import Curl
return Curl(uri, *args, **kwargs)
case 'sops':
from .ec.secret_store.Sops import Sops
return Sops(uri, *args, **kwargs)
case 'targpg':
from .ec.secret_store.TarGpg import TarGpg

410
src/python/jw/pkg/lib/ec/secret_store/Sops.py Normal file
View file

@ -0,0 +1,410 @@
from __future__ import annotations
import errno
import hashlib
import os
from contextlib import suppress
from typing import TYPE_CHECKING, Iterable, override
if TYPE_CHECKING:
from typing import Any
from ...ExecContext import ExecContext
from ...Uri import Uri
from ...base import StatResult
from ...file_types import FileType
from ...file_types import FT_DIR
from ...kv import parse_kv, serialize_kv
from ...Result import Result
from .Base import Base
from .layout import FT_SECRET, FT_SECRET_FILE, FT_SECRET_KEY
def _sops_dotenv_norm(content: bytes) -> bytes:
# -- The form sops' dotenv serialization produces: blank
# lines dropped, a single trailing newline, everything
# else verbatim. The identity for content already in
# that form
if not content:
return content
return b'\n'.join(line for line in content.split(b'\n') if line) + b'\n'
class SopsKeyError(Exception):
"""The age key file of the sops store could not be resolved"""
class Sops(Base):
"""A secret store of SOPS-encrypted files in a canonical layout.
The repository location is deployment information, not part of
the address: JW_SECRETS_SOPS_REPO names a local directory (used
as-is) or a git URL (cloned into a temporary directory until
close()), and JW_SECRETS_SOPS_ROOT an optional base directory
within it. The store URI addresses a subtree of the namespace,
as in the base class.
A store path names a file, and optionally a key below an entry.
The file of an entry is its address, and its keys are
addressable below it; a blob is a file at its address holding
non-key-value content. The store does not identify content by
file name: a file that decrypts to a key-value file is an entry,
anything else a blob. The repository files are sops files in
dotenv format: an entry's keys are its data lines, a blob is
stored under a single data key.
put() writes the file at its address, sops-encrypting the
content for its type: an entry wholesale, a key merged into its
entry, a blob as raw bytes. The store accepts entry, key, and
blob content only; the encryption recipient is derived from the
SOPS age key, and committing the repository is the caller's
step.
Files are decrypted with sops in a local exec context, and the
age key file is resolved from JW_SECRETS_SOPS_KEY_PATH, so the
key and the decrypted material stay on this machine.
"""
def __init__(self, uri: str | Uri, *args: Any, **kwargs: Any) -> None:
super().__init__(uri, *args, **kwargs)
self.__repo = os.environ.get('JW_SECRETS_SOPS_REPO', '.')
self.__base = os.environ.get('JW_SECRETS_SOPS_ROOT', '').strip('/')
self.__kv_cache: dict[str, dict[str, str]] = {}
self.__decrypt_cache: dict[bytes, bytes] = {}
self.___keyfile: str | None = None
self.___recipient: str | None = None
self.___git_checkout: str | None = None
self.___crypt_ec: ExecContext | None = None
async def __crypt_ec(self) -> ExecContext:
if self.___crypt_ec is None:
from ..Local import Local
# -- Security: decrypt in a local exec context for now so
# that the SOPS key and the decrypted material never
# leave this machine
self.___crypt_ec = Local()
return self.___crypt_ec
async def __git_checkout(self) -> str:
if self.___git_checkout is None:
from ...base import InputMode
ec = await self.__crypt_ec()
if os.path.isdir(self.__repo):
self.___git_checkout = self.__repo
else:
ret = await ec.mktemp('/tmp/jw-pkg-sops-XXXXXX', directory = True)
try:
await ec.run(
['git', 'clone', '--quiet', self.__repo, ret],
cmd_input = InputMode.NonInteractive,
throw = True,
)
except Exception:
await ec.erase(ret)
raise
self.___git_checkout = ret
return self.___git_checkout
async def __git_subdir(self, path: str) -> str:
checkout = await self.__git_checkout()
base = f'{checkout}/{self.__base}' if self.__base else checkout
return f'{base}/{path}' if path else base
async def __git_get(self, path: str) -> Result:
if path == '' or os.path.isdir(await self.__git_subdir(path)):
raise IsADirectoryError(path)
file_path = await self.__git_subdir(path)
if await self.__is_entry(path):
# -- The entry file is the serialized entry
return Result(stdout = await self.__decrypt(file_path))
if '/' in path:
entry, key = path.rsplit('/', 1)
if await self.__is_entry(entry):
kv = await self.__decrypt_kv_content(entry)
try:
return Result(stdout = kv[key].encode())
except KeyError:
raise FileNotFoundError(path) from None
if os.path.isfile(file_path):
from ...base import InputMode
ec = await self.__crypt_ec()
result = await ec.run(
[
'sops',
'--decrypt',
'--input-type',
'dotenv',
'--output-type',
'binary',
file_path,
],
cmd_input = InputMode.NonInteractive,
mod_env = {'SOPS_AGE_KEY_FILE': await self.__keyfile()},
throw = True,
)
return Result(stdout = result.stdout)
raise FileNotFoundError(path)
async def __git_readdir(self, path: str) -> Iterable[str]:
def __names(d: str) -> set[str]:
# -- No name filter: every file is store content
return set(os.listdir(d))
if await self.__is_entry(path):
return (await self.__decrypt_kv_content(path)).keys()
os_path = await self.__git_subdir(path)
if os.path.isfile(os_path):
raise NotADirectoryError(path)
if os.path.isdir(os_path):
names = __names(os_path)
default_path = self._default_path_in_storage(path)
if default_path is not None:
d = await self.__git_subdir(default_path)
if os.path.isdir(d):
names |= __names(d)
return sorted(names)
raise FileNotFoundError(path)
async def __git_stat(
self,
storage_path: str,
follow_symlinks: bool,
) -> StatResult:
stat_fn = os.lstat if not follow_symlinks else os.stat
file_path = await self.__git_subdir(storage_path)
if await self.__is_entry(storage_path):
return self._stat_result_from_os(stat_fn(file_path), FT_SECRET)
entry, sep, key = storage_path.rpartition('/')
if sep and key and await self.__is_entry(entry):
if key in await self.__decrypt_kv_content(entry):
# -- A key has no file of its own: the size stays 0
st = stat_fn(await self.__git_subdir(entry))
return self._stat_result_from_os(st, FT_SECRET_KEY, size = 0)
if os.path.isfile(file_path):
return self._stat_result_from_os(stat_fn(file_path), FT_SECRET_FILE)
if os.path.isdir(file_path):
return self._stat_result_from_os(stat_fn(file_path), FT_DIR)
raise FileNotFoundError(errno.ENOENT, f'stat: {storage_path}')
async def __keyfile(self) -> str:
# -- The house variable is the sole input; SOPS_AGE_KEY_FILE
# is set per invocation on the sops subprocess, never read
# from the process environment
if self.___keyfile is None:
keyfile = os.environ.get('JW_SECRETS_SOPS_KEY_PATH')
if not keyfile:
raise SopsKeyError('JW_SECRETS_SOPS_KEY_PATH is not set')
self.___keyfile = keyfile
return self.___keyfile
async def __recipient(self) -> str:
if self.___recipient is None:
from ...base import InputMode
keyfile = await self.__keyfile()
ec = await self.__crypt_ec()
# -- Pipe the output: in interactive mode the pty relays
# the child's stdout to the console as well
result = await ec.run(
['age-keygen', '-y', keyfile],
cmd_input = InputMode.NonInteractive,
throw = True,
)
self.___recipient = result.stdout.decode().strip()
return self.___recipient
async def __put_key(self, entry: str, key: str, content: bytes) -> Result:
# -- Merge the key into the entry, replacing the entry's
# existing value for the key
self.__kv_cache.pop(entry, None)
try:
kv = await self.__decrypt_kv_content(entry)
except FileNotFoundError:
kv = {}
kv[key] = content.decode()
await self.__encrypt(await self.__git_subdir(entry), serialize_kv(kv), 'dotenv')
return Result()
async def __put_entry(self, entry: str, content: bytes) -> Result:
# -- Replace the entry: the content is the serialized
# entry, encrypted wholesale, so a key not in it is gone
self.__kv_cache.pop(entry, None)
await self.__encrypt(await self.__git_subdir(entry), content, 'dotenv')
return Result()
async def __put_blob(self, path: str, content: bytes) -> Result:
await self.__encrypt(await self.__git_subdir(path), content, 'binary')
return Result()
async def __encrypt(self, path: str, content: bytes, input_type: str) -> None:
# -- Encrypt the content to the repository file. The
# plaintext lives in a temporary file next to it until the
# encryption succeeds, and the recipient is derived from
# the SOPS age key the store decrypts with
from ...base import InputMode
ec = await self.__crypt_ec()
recipient = await self.__recipient()
os.makedirs(os.path.dirname(path), exist_ok = True)
tmp = f'{path}.tmp'
try:
with open(tmp, 'wb') as f:
f.write(content)
await ec.run(
[
'sops',
'--input-type',
input_type,
'--output-type',
'dotenv',
'--encrypt',
'--age',
recipient,
'--output',
path,
tmp,
],
cmd_input = InputMode.NonInteractive,
mod_env = {'SOPS_AGE_KEY_FILE': await self.__keyfile()},
throw = True,
)
except Exception:
with suppress(FileNotFoundError):
os.unlink(tmp)
raise
os.unlink(tmp)
async def __decrypt(self, path: str) -> bytes:
# -- The decryption is a pure function of the file's content,
# so the result is cached by the content's hash: the entry
# check and the content fetch of one get() share one
# decryption, a repeated read of an unchanged file is
# free, and a changed file - after a put, or an external
# update of the store - re-decrypts under its new hash
# without any invalidation
if not os.path.isfile(path):
raise FileNotFoundError(path)
with open(path, 'rb') as f:
raw = f.read()
key = hashlib.sha256(raw).digest()
ret = self.__decrypt_cache.get(key)
if ret is None:
ec = await self.__crypt_ec()
ret = (
await ec.run(
[
'sops',
'decrypt',
'--input-type',
'dotenv',
'--output-type',
'dotenv',
],
cmd_input = raw,
mod_env = {'SOPS_AGE_KEY_FILE': await self.__keyfile()},
throw = True,
)
).stdout or b''
self.__decrypt_cache[key] = ret
return ret
async def __decrypt_kv_content(self, entry: str) -> dict[str, str]:
ret = self.__kv_cache.get(entry)
if ret is None:
path = await self.__git_subdir(entry)
ret = parse_kv(await self.__decrypt(path))
self.__kv_cache[entry] = ret
return ret
async def __is_entry(self, entry: str) -> bool:
# -- The file at entry is an entry: it decrypts to a kv
# file, and the single data key sops gives a blob is not
# all of its content
path = await self.__git_subdir(entry)
if not os.path.isfile(path):
return False
try:
kv = await self.__decrypt_kv_content(entry)
except SopsKeyError:
raise
except Exception:
return False
return not (len(kv) == 1 and 'data' in kv)
# -- Override base class methods
@override
async def _close(self) -> None:
if self.___git_checkout is not None and self.___git_checkout != self.__repo:
ec = await self.__crypt_ec()
await ec.erase(self.___git_checkout)
self.___git_checkout = None
@override
async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
path = self._unchroot(path)
return await self._call_with_fallback(
path,
lambda p: self.__git_stat(p, follow_symlinks),
)
@override
async def _get(
self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str
) -> Result:
path = self._unchroot(path)
return await self._call_with_fallback(path, self.__git_get)
@override
async def _put(
self,
path: str,
content: bytes,
wd: str | None,
throw: bool,
verbose: bool | None,
title: str,
owner: str | None,
group: str | None,
mode: str | None,
atomic: bool,
ftype: FileType,
) -> Result:
# -- Resolve the address to the absolute namespace path in
# the host tree; a write does not take the default
# fallback.
path = self._unchroot(path)
if not path:
raise ValueError(f'put: "{path}" is not an entry/key or blob path')
storage_path = self._path_in_storage(path)
if ftype.is_subtype(FT_SECRET):
return await self.__put_entry(storage_path, content)
if ftype.is_subtype(FT_SECRET_KEY):
entry, sep, key = storage_path.rpartition('/')
if not sep or not key:
raise ValueError(f'put: "{path}" is not of the form path/key')
return await self.__put_key(entry, key, content)
if ftype.is_subtype(FT_SECRET_FILE):
return await self.__put_blob(storage_path, content)
raise ValueError(f'put: ftype {ftype} is not a store content type')
@override
def _put_is_noop(self, existing: bytes, content: bytes, ftype: FileType) -> bool:
# -- sops re-serializes a dotenv entry on decrypt: it
# drops blank lines and adds a trailing newline, so
# the byte check of Base._put_is_noop() never fires
# for content outside that form, and every put
# re-encrypts the unchanged entry. Compare the
# sops-canonical forms. A blob round-trips
# byte-exact and keeps the byte check
if ftype.is_subtype(FT_SECRET):
return _sops_dotenv_norm(existing) == _sops_dotenv_norm(content)
return existing == content
@override
async def _readdir(self, path: str) -> Iterable[str]:
path = self._unchroot(path)
return await self._call_with_fallback(path, self.__git_readdir)