mirror of
https://github.com/borgmatic-collective/borgmatic.git
synced 2026-08-01 22:23:01 +02:00
Add credential loading from file, KeePassXC, and Docker/Podman secrets.
This commit is contained in:
@@ -1,3 +1,6 @@
|
||||
1.9.11
|
||||
* Add credential loading from file, KeePassXC, and Docker/Podman secrets. See the documentation for
|
||||
more information: https://torsion.org/borgmatic/docs/how-to/provide-your-passwords/
|
||||
1.9.10
|
||||
* #966: Add a "{credential ...}" syntax for loading systemd credentials into borgmatic
|
||||
configuration files. See the documentation for more information:
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
SECRET_NAME_PATTERN = re.compile(r'^\w+$')
|
||||
SECRETS_DIRECTORY = '/run/secrets'
|
||||
|
||||
|
||||
def load_credential(hook_config, config, credential_parameters):
|
||||
'''
|
||||
Given the hook configuration dict, the configuration dict, and a credential parameters tuple
|
||||
containing a secret name to load, read the secret from the corresponding container secrets file
|
||||
and return it.
|
||||
|
||||
Raise ValueError if the credential parameters is not one element, the secret name is invalid, or
|
||||
the secret file cannot be read.
|
||||
'''
|
||||
try:
|
||||
(secret_name,) = credential_parameters
|
||||
except ValueError:
|
||||
raise ValueError(f'Cannot load invalid secret name: "{' '.join(credential_parameters)}"')
|
||||
|
||||
if not SECRET_NAME_PATTERN.match(secret_name):
|
||||
raise ValueError(f'Cannot load invalid secret name: "{secret_name}"')
|
||||
|
||||
try:
|
||||
with open(os.path.join(SECRETS_DIRECTORY, secret_name)) as secret_file:
|
||||
return secret_file.read().rstrip(os.linesep)
|
||||
except (FileNotFoundError, OSError) as error:
|
||||
logger.warning(error)
|
||||
|
||||
raise ValueError(f'Cannot load secret "{secret_name}" from file: {error.filename}')
|
||||
@@ -0,0 +1,29 @@
|
||||
import logging
|
||||
import os
|
||||
import shlex
|
||||
|
||||
import borgmatic.execute
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def load_credential(hook_config, config, credential_parameters):
|
||||
'''
|
||||
Given the hook configuration dict, the configuration dict, and a credential parameters tuple
|
||||
containing a credential path to load, load the credential from file and return it.
|
||||
|
||||
Raise ValueError if the credential parameters is not one element or the secret file cannot be
|
||||
read.
|
||||
'''
|
||||
try:
|
||||
(credential_path,) = credential_parameters
|
||||
except ValueError:
|
||||
raise ValueError(f'Cannot load credential with invalid credential path: "{' '.join(credential_parameters)}"')
|
||||
|
||||
try:
|
||||
with open(credential_path) as credential_file:
|
||||
return credential_file.read().rstrip(os.linesep)
|
||||
except (FileNotFoundError, OSError) as error:
|
||||
logger.warning(error)
|
||||
|
||||
raise ValueError(f'Cannot load credential "{credential_name}" from file: {error.filename}')
|
||||
@@ -0,0 +1,36 @@
|
||||
import logging
|
||||
import os
|
||||
import shlex
|
||||
|
||||
import borgmatic.execute
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def load_credential(hook_config, config, credential_parameters):
|
||||
'''
|
||||
Given the hook configuration dict, the configuration dict, and a credential parameters tuple
|
||||
containing a KeePassXC database path and an attribute name to load, run keepassxc-cli to fetch
|
||||
the corresponidng KeePassXC credential and return it.
|
||||
|
||||
Raise ValueError if keepassxc-cli can't retrieve the credential.
|
||||
'''
|
||||
try:
|
||||
(database_path, attribute_name) = credential_parameters
|
||||
except ValueError:
|
||||
raise ValueError(f'Cannot load credential with invalid KeePassXC database path and attribute name: "{' '.join(credential_parameters)}"')
|
||||
|
||||
if not os.path.exists(database_path):
|
||||
raise ValueError(f'Cannot load credential because KeePassXC database path does not exist: {database_path}')
|
||||
|
||||
return borgmatic.execute.execute_command_and_capture_output(
|
||||
(
|
||||
'keepassxc-cli',
|
||||
'show',
|
||||
'--show-protected',
|
||||
'--attributes',
|
||||
'Password',
|
||||
database_path,
|
||||
attribute_name,
|
||||
)
|
||||
).rstrip(os.linesep)
|
||||
@@ -1,5 +1,6 @@
|
||||
import functools
|
||||
import re
|
||||
import shlex
|
||||
|
||||
import borgmatic.hooks.dispatch
|
||||
|
||||
@@ -7,11 +8,9 @@ IS_A_HOOK = False
|
||||
|
||||
|
||||
CREDENTIAL_PATTERN = re.compile(
|
||||
r'\{credential +(?P<hook_name>[A-Za-z0-9_]+) +(?P<credential_name>[A-Za-z0-9_]+)\}'
|
||||
r'\{credential( +(?P<contents>.*))?\}'
|
||||
)
|
||||
|
||||
GENERAL_CREDENTIAL_PATTERN = re.compile(r'\{credential( +[^}]*)?\}')
|
||||
|
||||
|
||||
@functools.cache
|
||||
def resolve_credential(value):
|
||||
@@ -27,16 +26,21 @@ def resolve_credential(value):
|
||||
if value is None:
|
||||
return value
|
||||
|
||||
result = CREDENTIAL_PATTERN.sub(
|
||||
lambda matcher: borgmatic.hooks.dispatch.call_hook(
|
||||
'load_credential', {}, matcher.group('hook_name'), matcher.group('credential_name')
|
||||
),
|
||||
value,
|
||||
)
|
||||
matcher = CREDENTIAL_PATTERN.match(value)
|
||||
|
||||
# If we've tried to parse the credential, but the parsed result still looks kind of like a
|
||||
# credential, it means it's invalid syntax.
|
||||
if GENERAL_CREDENTIAL_PATTERN.match(result):
|
||||
if not matcher:
|
||||
return value
|
||||
|
||||
contents = matcher.group('contents')
|
||||
|
||||
if not contents:
|
||||
raise ValueError(f'Cannot load credential with invalid syntax "{value}"')
|
||||
|
||||
return result
|
||||
(hook_name, *credential_parameters) = shlex.split(contents)
|
||||
|
||||
if not credential_parameters:
|
||||
raise ValueError(f'Cannot load credential with invalid syntax "{value}"')
|
||||
|
||||
return borgmatic.hooks.dispatch.call_hook(
|
||||
'load_credential', {}, hook_name, credential_parameters
|
||||
)
|
||||
|
||||
@@ -5,29 +5,29 @@ import re
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
CREDENTIAL_NAME_PATTERN = re.compile(r'^\w+$')
|
||||
SECRET_NAME_PATTERN = re.compile(r'^\w+$')
|
||||
SECRETS_DIRECTORY = '/run/secrets'
|
||||
|
||||
|
||||
def load_credential(hook_config, config, credential_name):
|
||||
def load_credential(hook_config, config, credential_parameters):
|
||||
'''
|
||||
Given the hook configuration dict, the configuration dict, and a credential name to load, read
|
||||
the credential from the corresponding systemd credential file and return it.
|
||||
Given the hook configuration dict, the configuration dict, and a credential parameters tuple
|
||||
containing a secret name to load, read the secret from the corresponding container secrets file
|
||||
and return it.
|
||||
|
||||
Raise ValueError if the systemd CREDENTIALS_DIRECTORY environment variable is not set, the
|
||||
credential name is invalid, or the credential file cannot be read.
|
||||
Raise ValueError if the credential parameters is not one element, the secret name is invalid, or
|
||||
the secret file cannot be read.
|
||||
'''
|
||||
credentials_directory = os.environ.get('CREDENTIALS_DIRECTORY')
|
||||
try:
|
||||
(secert_name,) = credential_parameters
|
||||
except ValueError:
|
||||
raise ValueError(f'Cannot load invalid secret name: "{' '.join(credential_parameters)}"')
|
||||
|
||||
if not credentials_directory:
|
||||
raise ValueError(
|
||||
f'Cannot load credential "{credential_name}" because the systemd CREDENTIALS_DIRECTORY environment variable is not set'
|
||||
)
|
||||
|
||||
if not CREDENTIAL_NAME_PATTERN.match(credential_name):
|
||||
raise ValueError(f'Cannot load invalid credential name "{credential_name}"')
|
||||
if not SECRET_NAME_PATTERN.match(SECRET_NAME):
|
||||
raise ValueError(f'Cannot load invalid secret name: "{credential_name}"')
|
||||
|
||||
try:
|
||||
with open(os.path.join(credentials_directory, credential_name)) as credential_file:
|
||||
with open(os.path.join(SECRETS_DIRECTORY, credential_name)) as credential_file:
|
||||
return credential_file.read().rstrip(os.linesep)
|
||||
except (FileNotFoundError, OSError) as error:
|
||||
logger.warning(error)
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "borgmatic"
|
||||
version = "1.9.10"
|
||||
version = "1.9.11"
|
||||
authors = [
|
||||
{ name="Dan Helfman", email="witten@torsion.org" },
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user