Coverage for app/venv/lib/python3.14/site-packages/weblate/utils/backup.py: 31%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5"""Backup automation based on borg.""" 

6 

7from __future__ import annotations 

8 

9import os 

10import string 

11import subprocess 

12from random import SystemRandom 

13from urllib.parse import urlparse 

14 

15from django.conf import settings 

16 

17from weblate.trans.util import get_clean_env 

18from weblate.utils.data import data_dir 

19from weblate.utils.errors import add_breadcrumb, report_error 

20from weblate.utils.files import cleanup_error_message 

21from weblate.utils.lock import WeblateLock 

22from weblate.vcs.ssh import SSH_WRAPPER, add_host_key 

23 

24CACHEDIR = """Signature: 8a477f597d28d172789f06886806bc55 

25# This file is a cache directory tag created by Weblate 

26# For information about cache directory tags, see: 

27# https://bford.info/cachedir/spec.html 

28""" 

29 

30 

31def ensure_backup_dir(): 

32 backup_dir = data_dir("backups") 

33 if not os.path.exists(backup_dir): 

34 os.makedirs(backup_dir) 

35 return backup_dir 

36 

37 

38def backup_lock(): 

39 backup_dir = ensure_backup_dir() 

40 return WeblateLock( 

41 lock_path=backup_dir, 

42 scope="backuplock", 

43 key=0, 

44 slug="", 

45 cache_template="lock:{scope}", 

46 file_template=".{scope}", 

47 timeout=120, 

48 ) 

49 

50 

51class BackupError(Exception): 

52 pass 

53 

54 

55def make_password(length: int = 50): 

56 generator = SystemRandom() 

57 chars = string.ascii_letters + string.digits + "!@#$%^&*()" 

58 return "".join(generator.choice(chars) for i in range(length)) 

59 

60 

61def tag_cache_dirs() -> None: 

62 """Create CACHEDIR.TAG in our cache dirs to exclude from backups.""" 

63 dirs = [ 

64 # Fontconfig cache 

65 data_dir("cache", "fonts"), 

66 # Static files (default is inside data) 

67 settings.STATIC_ROOT, 

68 # Project backups 

69 data_dir("projectbackups"), 

70 ] 

71 # Django file based caches 

72 dirs.extend( 

73 cache["LOCATION"] 

74 for cache in settings.CACHES.values() 

75 if cache["BACKEND"] == "django.core.cache.backends.filebased.FileBasedCache" 

76 ) 

77 

78 # Create CACHEDIR.TAG in each cache dir 

79 for name in dirs: 

80 tagfile = os.path.join(name, "CACHEDIR.TAG") 

81 if os.path.exists(name) and not os.path.exists(tagfile): 

82 with open(tagfile, "w") as handle: 

83 handle.write(CACHEDIR) 

84 

85 

86def run_borg(cmd: list[str], env: dict[str, str] | None = None) -> str: 

87 """Execute borgbackup.""" 

88 with backup_lock(): 

89 SSH_WRAPPER.create() 

90 try: 

91 return subprocess.check_output( 

92 ["borg", "--rsh", SSH_WRAPPER.filename, *cmd], 

93 stderr=subprocess.STDOUT, 

94 env=get_clean_env(env), 

95 text=True, 

96 ) 

97 except OSError as error: 

98 report_error("Borg could not be executed") 

99 msg = f"Could not execute borg program: {error}" 

100 raise BackupError(msg) from error 

101 except subprocess.CalledProcessError as error: 

102 add_breadcrumb( 

103 category="backup", message="borg output", stdout=error.stdout 

104 ) 

105 report_error("Borg failed") 

106 raise BackupError(cleanup_error_message(error.stdout)) from error 

107 

108 

109def initialize(location: str, passphrase: str) -> str: 

110 """Initialize repository.""" 

111 parsed = urlparse(location) 

112 if parsed.hostname: 

113 add_host_key(None, parsed.hostname, parsed.port) 

114 return run_borg( 

115 ["init", "--encryption", "repokey-blake2", location], 

116 {"BORG_NEW_PASSPHRASE": passphrase}, 

117 ) 

118 

119 

120def get_paper_key(location: str) -> str: 

121 """Get paper key for recovery.""" 

122 return run_borg(["key", "export", "--paper", location]) 

123 

124 

125def backup(location: str, passphrase: str) -> str: 

126 """Perform DATA_DIR backup.""" 

127 tag_cache_dirs() 

128 command = [ 

129 "create", 

130 "--verbose", 

131 "--list", 

132 "--filter", 

133 "AME", 

134 "--stats", 

135 "--exclude-caches", 

136 "--exclude", 

137 "*/.config/borg", 

138 "--exclude", 

139 "lost+found", 

140 "--compression", 

141 "auto,zstd", 

142 ] 

143 if settings.BORG_EXTRA_ARGS: 

144 command.extend(settings.BORG_EXTRA_ARGS) 

145 command.extend( 

146 [ 

147 f"{location}::{{now}}", 

148 settings.DATA_DIR, 

149 ], 

150 ) 

151 return run_borg( 

152 command, 

153 {"BORG_PASSPHRASE": passphrase}, 

154 ) 

155 

156 

157def prune(location: str, passphrase: str) -> str: 

158 """Prune past backups.""" 

159 return run_borg( 

160 [ 

161 "prune", 

162 "--list", 

163 "--keep-within", 

164 "2d", 

165 "--keep-daily", 

166 "14", 

167 "--keep-weekly", 

168 "8", 

169 "--keep-monthly", 

170 "6", 

171 location, 

172 ], 

173 {"BORG_PASSPHRASE": passphrase}, 

174 ) 

175 

176 

177def cleanup(location: str, passphrase: str, initial: bool) -> str: 

178 cmd = ["compact"] 

179 if initial: 

180 cmd.append("--cleanup-commits") 

181 cmd.append(location) 

182 return run_borg(cmd, {"BORG_PASSPHRASE": passphrase})