2
0

paths.py 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170
  1. import logging
  2. import os
  3. import tempfile
  4. logger = logging.getLogger(__name__)
  5. def expand_user_in_path(path):
  6. '''
  7. Given a directory path, expand any tildes in it.
  8. '''
  9. try:
  10. return os.path.expanduser(path or '') or None
  11. except TypeError:
  12. return None
  13. def get_working_directory(config): # pragma: no cover
  14. '''
  15. Given a configuration dict, get the working directory from it, expanding any tildes.
  16. '''
  17. return expand_user_in_path(config.get('working_directory'))
  18. def get_borgmatic_source_directory(config):
  19. '''
  20. Given a configuration dict, get the (deprecated) borgmatic source directory, expanding any
  21. tildes. Defaults to ~/.borgmatic.
  22. '''
  23. return expand_user_in_path(config.get('borgmatic_source_directory') or '~/.borgmatic')
  24. TEMPORARY_DIRECTORY_PREFIX = 'borgmatic-'
  25. def replace_temporary_subdirectory_with_glob(path):
  26. '''
  27. Given an absolute temporary directory path, look for a subdirectory within it starting with the
  28. temporary directory prefix and replace it with an appropriate glob. For instance, given:
  29. /tmp/borgmatic-aet8kn93/borgmatic
  30. ... replace it with:
  31. /tmp/borgmatic-*/borgmatic
  32. This is useful for finding previous temporary directories from prior borgmatic runs.
  33. '''
  34. return os.path.join(
  35. '/',
  36. *(
  37. (
  38. f'{TEMPORARY_DIRECTORY_PREFIX}*'
  39. if subdirectory.startswith(TEMPORARY_DIRECTORY_PREFIX)
  40. else subdirectory
  41. )
  42. for subdirectory in path.split(os.path.sep)
  43. ),
  44. )
  45. class Runtime_directory:
  46. '''
  47. A Python context manager for creating and cleaning up the borgmatic runtime directory used for
  48. storing temporary runtime data like streaming database dumps and bootstrap metadata.
  49. Example use as a context manager:
  50. with borgmatic.config.paths.Runtime_directory(config) as borgmatic_runtime_directory:
  51. do_something_with(borgmatic_runtime_directory)
  52. For the scope of that "with" statement, the runtime directory is available. Afterwards, it
  53. automatically gets cleaned up as necessary.
  54. '''
  55. def __init__(self, config, log_prefix):
  56. '''
  57. Given a configuration dict and a log prefix, determine the borgmatic runtime directory,
  58. creating a secure, temporary directory within it if necessary. Defaults to
  59. $XDG_RUNTIME_DIR/./borgmatic or $RUNTIME_DIRECTORY/./borgmatic or
  60. $TMPDIR/borgmatic-[random]/./borgmatic or $TEMP/borgmatic-[random]/./borgmatic or
  61. /tmp/borgmatic-[random]/./borgmatic where "[random]" is a randomly generated string intended
  62. to avoid path collisions.
  63. If XDG_RUNTIME_DIR or RUNTIME_DIRECTORY is set and already ends in "/borgmatic", then don't
  64. tack on a second "/borgmatic" path component.
  65. The "/./" is taking advantage of a Borg feature such that the part of the path before the "/./"
  66. does not get stored in the file path within an archive. That way, the path of the runtime
  67. directory can change without leaving database dumps within an archive inaccessible.
  68. '''
  69. runtime_directory = (
  70. config.get('user_runtime_directory')
  71. or os.environ.get('XDG_RUNTIME_DIR') # Set by PAM on Linux.
  72. or os.environ.get('RUNTIME_DIRECTORY') # Set by systemd if configured.
  73. )
  74. if runtime_directory:
  75. self.temporary_directory = None
  76. else:
  77. base_directory = os.environ.get('TMPDIR') or os.environ.get('TEMP') or '/tmp'
  78. os.makedirs(base_directory, mode=0o700, exist_ok=True)
  79. self.temporary_directory = tempfile.TemporaryDirectory(
  80. prefix=TEMPORARY_DIRECTORY_PREFIX,
  81. dir=base_directory,
  82. )
  83. runtime_directory = self.temporary_directory.name
  84. (base_path, final_directory) = os.path.split(runtime_directory.rstrip(os.path.sep))
  85. self.runtime_path = expand_user_in_path(
  86. os.path.join(
  87. base_path if final_directory == 'borgmatic' else runtime_directory,
  88. '.', # Borg 1.4+ "slashdot" hack.
  89. 'borgmatic',
  90. )
  91. )
  92. os.makedirs(self.runtime_path, mode=0o700, exist_ok=True)
  93. logger.debug(f'{log_prefix}: Using runtime directory {os.path.normpath(self.runtime_path)}')
  94. def __enter__(self):
  95. '''
  96. Return the borgmatic runtime path as a string.
  97. '''
  98. return self.runtime_path
  99. def __exit__(self, exception, value, traceback):
  100. '''
  101. Delete any temporary directory that was created as part of initialization.
  102. '''
  103. if self.temporary_directory:
  104. try:
  105. self.temporary_directory.cleanup()
  106. # The cleanup() call errors if, for instance, there's still a
  107. # mounted filesystem within the temporary directory. There's
  108. # nothing we can do about that here, so swallow the error.
  109. except OSError:
  110. pass
  111. def make_runtime_directory_glob(borgmatic_runtime_directory):
  112. '''
  113. Given a borgmatic runtime directory path, make a glob that would match that path, specifically
  114. replacing any randomly generated temporary subdirectory with "*" since such a directory's name
  115. changes on every borgmatic run.
  116. '''
  117. return os.path.join(
  118. *(
  119. '*' if subdirectory.startswith(TEMPORARY_DIRECTORY_PREFIX) else subdirectory
  120. for subdirectory in os.path.normpath(borgmatic_runtime_directory).split(os.path.sep)
  121. )
  122. )
  123. def get_borgmatic_state_directory(config):
  124. '''
  125. Given a configuration dict, get the borgmatic state directory used for storing borgmatic state
  126. files like records of when checks last ran. Defaults to $XDG_STATE_HOME/borgmatic or
  127. ~/.local/state/./borgmatic.
  128. '''
  129. return expand_user_in_path(
  130. os.path.join(
  131. config.get('user_state_directory')
  132. or os.environ.get('XDG_STATE_HOME')
  133. or os.environ.get('STATE_DIRECTORY') # Set by systemd if configured.
  134. or '~/.local/state',
  135. 'borgmatic',
  136. )
  137. )