← Back to branch summary

~abentley/simplestreams/fix-read-signed-no-check

« back to all changes in this revision

Viewing changes to tests/unittests/test_badmirrors.py

  • Committer: Scott Moser
  • Date: 2015-09-22 20:28:53 UTC
  • mfrom: (398.1.21 trunk.lp1487004)
  • Revision ID: smoser@ubuntu.com-20150922202853-hxqmj6bpif5n62v9
https://launchpad.net/bugs/1487004
provide insert_item with a contentsource that does checksumming

Previously, insert_item would receive a 'contentsource' (essentially
a file like object to be read as a stream).  The user of that object
needed to calculate checksums and verify the data they read.

Now, instead the contentsource will do checksumming as read() 
operations are done, and will raise a checksum error in any failure
case.

Thus to use this, the user now simply has to loop over reads
and catch the exception.

stream data is now expected to have valid checksums and size on
all items with a path.  If the user is using a stream source that
does not have either size or checksum information, they have a few
options:
 a.) [legacy/SRU only] set environment variable
     SS_MISSING_ITEM_CHECKSUM_BEHAVIOR can be set to
        silent: behave exactly as before.  No checksumming is done,
                no warnings are emitted.  The consumer of the
                contentsource must check checksums.
        warn:   log messages at WARN level (same as default/unset)
        fail:   the new behavior. raise an InvalidChecksum exception.

 b.) instantiate the BasicMirrorWriter with config checksumming_reader=False
     the default for that config setting is True, meaning that you
     will get a reader that checksums content as it goes and 
     raises exception on bad data.

 c.) fix their source to have a sha256sum and a size


The 'sstream-mirror' program now has a '--no-checksumming-reader' flag
that does 'b' for this mirror.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/unittests/test_badmirrors.py
 
1
from unittest import TestCase
 
2
from tests.testutil import get_mirror_reader
 
3
from simplestreams.mirrors import (
 
4
    ObjectStoreMirrorWriter, ObjectStoreMirrorReader)
 
5
from simplestreams.objectstores import MemoryObjectStore
 
6
from simplestreams import util
 
7
from simplestreams import checksum_util
 
8
from simplestreams import mirrors
 
9
 
 
10
 
 
11
class TestBadDataSources(TestCase):
 
12
    """Test of Bad Data in a datasource."""
 
13
 
 
14
    dlpath = "streams/v1/download.json"
 
15
    pedigree = ("com.example:product1", "20150915", "item1")
 
16
    item_path = "product1/20150915/text.txt"
 
17
    example = "minimal"
 
18
 
 
19
    def setUp(self):
 
20
        self.src = self.get_clean_src(self.example, path=self.dlpath)
 
21
        self.target = ObjectStoreMirrorWriter(
 
22
            config={}, objectstore=MemoryObjectStore())
 
23
 
 
24
    def get_clean_src(self, exname, path):
 
25
        good_src = get_mirror_reader(exname)
 
26
        objectstore = MemoryObjectStore(None)
 
27
        target = ObjectStoreMirrorWriter(config={}, objectstore=objectstore)
 
28
        target.sync(good_src, path)
 
29
 
 
30
        # clean the .data out of the mirror so it doesn't get read
 
31
        keys = list(objectstore.data.keys())
 
32
        for k in keys:
 
33
            if k.startswith(".data"):
 
34
                del objectstore.data[k]
 
35
 
 
36
        return ObjectStoreMirrorReader(
 
37
            objectstore=objectstore, policy=lambda content, path: content)
 
38
 
 
39
    def test_sanity_valid(self):
 
40
        # verify that the tests are fine on expected pass
 
41
        _moditem(self.src, self.dlpath, self.pedigree, lambda c: c)
 
42
        self.target.sync(self.src, self.dlpath)
 
43
 
 
44
    def test_larger_size_causes_bad_checksum(self):
 
45
        def size_plus_1(item):
 
46
            item['size'] = int(item['size']) + 1
 
47
            return item
 
48
 
 
49
        _moditem(self.src, self.dlpath, self.pedigree, size_plus_1)
 
50
        self.assertRaises(checksum_util.InvalidChecksum,
 
51
                          self.target.sync, self.src, self.dlpath)
 
52
 
 
53
    def test_smaller_size_causes_bad_checksum(self):
 
54
        def size_minus_1(item):
 
55
            item['size'] = int(item['size']) - 1
 
56
            return item
 
57
        _moditem(self.src, self.dlpath, self.pedigree, size_minus_1)
 
58
        self.assertRaises(checksum_util.InvalidChecksum,
 
59
                          self.target.sync, self.src, self.dlpath)
 
60
 
 
61
    def test_too_much_content_causes_bad_checksum(self):
 
62
        self.src.objectstore.data[self.item_path] += b"extra"
 
63
        self.assertRaises(checksum_util.InvalidChecksum,
 
64
                          self.target.sync, self.src, self.dlpath)
 
65
 
 
66
    def test_too_little_content_causes_bad_checksum(self):
 
67
        orig = self.src.objectstore.data[self.item_path]
 
68
        self.src.objectstore.data[self.item_path] = orig[0:-1]
 
69
        self.assertRaises(checksum_util.InvalidChecksum,
 
70
                          self.target.sync, self.src, self.dlpath)
 
71
 
 
72
    def test_busted_checksum_causes_bad_checksum(self):
 
73
        def break_checksum(item):
 
74
            chars = "0123456789abcdef"
 
75
            orig = item['sha256']
 
76
            item['sha256'] = ''.join(
 
77
                [chars[(chars.find(c) + 1) % len(chars)] for c in orig])
 
78
            return item
 
79
 
 
80
        _moditem(self.src, self.dlpath, self.pedigree, break_checksum)
 
81
        self.assertRaises(checksum_util.InvalidChecksum,
 
82
                          self.target.sync, self.src, self.dlpath)
 
83
 
 
84
    def test_changed_content_causes_bad_checksum(self):
 
85
        # correct size but different content should raise bad checksum
 
86
        self.src.objectstore.data[self.item_path] = ''.join(
 
87
            ["x" for c in self.src.objectstore.data[self.item_path]])
 
88
        self.assertRaises(checksum_util.InvalidChecksum,
 
89
                          self.target.sync, self.src, self.dlpath)
 
90
 
 
91
    def test_no_checksums_cause_bad_checksum(self):
 
92
        def del_checksums(item):
 
93
            for c in checksum_util.item_checksums(item).keys():
 
94
                del item[c]
 
95
            return item
 
96
 
 
97
        _moditem(self.src, self.dlpath, self.pedigree, del_checksums)
 
98
        with _patched_missing_sum("fail"):
 
99
            self.assertRaises(checksum_util.InvalidChecksum,
 
100
                              self.target.sync, self.src, self.dlpath)
 
101
 
 
102
    def test_missing_size_causes_bad_checksum(self):
 
103
        def del_size(item):
 
104
            del item['size']
 
105
            return item
 
106
 
 
107
        _moditem(self.src, self.dlpath, self.pedigree, del_size)
 
108
        with _patched_missing_sum("fail"):
 
109
            self.assertRaises(checksum_util.InvalidChecksum,
 
110
                              self.target.sync, self.src, self.dlpath)
 
111
 
 
112
 
 
113
class _patched_missing_sum(object):
 
114
    """This patches the legacy mode for missing checksum info so
 
115
    that it behaves like the new code path.  Thus we can make
 
116
    the test run correctly"""
 
117
    def __init__(self, mode="fail"):
 
118
        self.mode = mode
 
119
 
 
120
    def __enter__(self):
 
121
        self.modmcb = getattr(mirrors, '_missing_cksum_behavior', {})
 
122
        self.orig = self.modmcb.copy()
 
123
        if self.modmcb:
 
124
            self.modmcb['mode'] = self.mode
 
125
        return self
 
126
 
 
127
    def __exit__(self, type, value, traceback):
 
128
        self.patch = self.orig
 
129
 
 
130
 
 
131
def _moditem(src, path, pedigree, modfunc):
 
132
    # load the products data at 'path' in 'src' mirror, then call modfunc
 
133
    # on the data found at pedigree. and store the updated data.
 
134
    sobj = src.objectstore
 
135
    tree = util.load_content(sobj.source(path).read())
 
136
    item = util.products_exdata(tree, pedigree, insert_fieldnames=False)
 
137
    util.products_set(tree, modfunc(item), pedigree)
 
138
    sobj.insert_content(path, util.dump_data(tree))