1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
#!/usr/bin/env python3
'''
t4_adm.py - this file is part of S3QL.
Copyright © 2008 Nikolaus Rath <Nikolaus@rath.org>
This work can be distributed under the terms of the GNU GPLv3.
'''
if __name__ == '__main__':
import pytest
import sys
sys.exit(pytest.main([__file__] + sys.argv[1:]))
from s3ql.backends import local
from s3ql.backends.comprenc import ComprencBackend
from s3ql.backends.common import CorruptedObjectError
from argparse import Namespace
import shutil
import re
import tempfile
import unittest
import subprocess
import pytest
import os
@pytest.mark.usefixtures('pass_s3ql_cmd_argv', 'pass_reg_output')
class AdmTests(unittest.TestCase):
def setUp(self):
self.cache_dir = tempfile.mkdtemp(prefix='s3ql-cache-')
self.backend_dir = tempfile.mkdtemp(prefix='s3ql-backend-')
self.storage_url = 'local://' + self.backend_dir
self.passphrase = 'oeut3d'
def tearDown(self):
shutil.rmtree(self.cache_dir)
shutil.rmtree(self.backend_dir)
def mkfs(self):
proc = subprocess.Popen(self.s3ql_cmd_argv('mkfs.s3ql') +
['-L', 'test fs', '--max-obj-size', '500',
'--authfile', '/dev/null', '--cachedir', self.cache_dir,
'--quiet', self.storage_url ],
stdin=subprocess.PIPE, universal_newlines=True,
stdout=subprocess.PIPE)
print(self.passphrase, file=proc.stdin)
print(self.passphrase, file=proc.stdin)
proc.stdin.close()
stdout = proc.stdout.read()
proc.stdout.close()
self.assertEqual(proc.wait(), 0)
self.reg_output(r'^WARNING: Maximum object sizes less than '
'1 MiB will degrade performance\.$', count=1)
return stdout
@unittest.skipUnless(os.path.exists('/proc/mounts'), '/proc/mounts not available')
def test_passphrase(self):
self.mkfs()
passphrase_new = 'sd982jhd'
proc = subprocess.Popen(self.s3ql_cmd_argv('s3qladm') +
[ '--quiet', '--log', 'none', '--authfile',
'/dev/null', 'passphrase', self.storage_url ],
stdin=subprocess.PIPE, universal_newlines=True)
print(self.passphrase, file=proc.stdin)
print(passphrase_new, file=proc.stdin)
print(passphrase_new, file=proc.stdin)
proc.stdin.close()
self.assertEqual(proc.wait(), 0)
plain_backend = local.Backend(Namespace(
storage_url=self.storage_url))
backend = ComprencBackend(passphrase_new.encode(), ('zlib', 6), plain_backend)
backend.fetch('s3ql_passphrase') # will fail with wrong pw
def test_clear(self):
self.mkfs()
proc = subprocess.Popen(self.s3ql_cmd_argv('s3qladm') +
[ '--quiet', '--log', 'none', '--authfile',
'/dev/null', 'clear', self.storage_url ],
stdin=subprocess.PIPE, universal_newlines=True)
print('yes', file=proc.stdin)
proc.stdin.close()
self.assertEqual(proc.wait(), 0)
plain_backend = local.Backend(Namespace(
storage_url=self.storage_url))
assert list(plain_backend.list()) == []
def test_key_recovery(self):
mkfs_output = self.mkfs()
hit = re.search(r'^---BEGIN MASTER KEY---\n'
r'(.+)\n'
r'---END MASTER KEY---$', mkfs_output, re.MULTILINE)
assert hit
master_key = hit.group(1)
plain_backend = local.Backend(Namespace(
storage_url=self.storage_url))
del plain_backend['s3ql_passphrase'] # Oops
backend = ComprencBackend(self.passphrase.encode(), ('zlib', 6), plain_backend)
with pytest.raises(CorruptedObjectError):
backend.fetch('s3ql_metadata')
passphrase_new = 'sd982jhd'
proc = subprocess.Popen(self.s3ql_cmd_argv('s3qladm') +
[ '--quiet', '--log', 'none', '--authfile',
'/dev/null', 'recover-key', self.storage_url ],
stdin=subprocess.PIPE, universal_newlines=True)
print(master_key, file=proc.stdin)
print(passphrase_new, file=proc.stdin)
print(passphrase_new, file=proc.stdin)
proc.stdin.close()
self.assertEqual(proc.wait(), 0)
backend = ComprencBackend(passphrase_new.encode(), ('zlib', 6), plain_backend)
backend.fetch('s3ql_passphrase') # will fail with wrong pw
|
