self._privkey = None
self._encprivkey = None
- # Starting with MDMF caps, we allowed arbitrary extensions in
- # caps. If we were initialized with a cap that had extensions,
- # we want to remember them so we can tell MutableFileVersions
- # about them.
- extensions = self._uri.get_extension_params()
- if extensions:
- extensions = map(int, extensions)
- suspected_k, suspected_segsize = extensions
- self._downloader_hints['k'] = suspected_k
- self._downloader_hints['segsize'] = suspected_segsize
-
return self
def create_with_keys(self, (pubkey, privkey), contents,
def set_downloader_hints(self, hints):
self._downloader_hints = hints
- extensions = [ hints["k"], hints["segsize"] ]
- self._uri.set_extension_params(extensions)
-
def _did_upload(self, res, size):
self._most_recent_size = size
if base32.could_be_base32_encoded(piece):
storage_index = base32.a2b(piece)
fingerprint = hashutil.ssk_pubkey_fingerprint_hash(pubkey)
- hints = [str(k), str(segsize)]
- u = MDMFVerifierURI(storage_index, fingerprint, hints)
+ u = MDMFVerifierURI(storage_index, fingerprint)
verify_cap = u.to_string()
print >>out, " verify-cap:", quote_output(verify_cap, quotemarks=False)
data = initial_contents.read(initial_contents.get_size())
data = "".join(data)
self.all_contents[self.storage_index] = data
- self.my_uri.set_extension_params([self._k, self._segsize])
return defer.succeed(self)
def _get_initial_contents(self, contents):
if contents is None:
new_data = new_contents.read(new_contents.get_size())
new_data = "".join(new_data)
self.all_contents[self.storage_index] = new_data
- self.my_uri.set_extension_params([self._k, self._segsize])
return defer.succeed(None)
def modify(self, modifier):
# this does not implement FileTooLargeError, but the real one does
old_contents = self.all_contents[self.storage_index]
new_data = modifier(old_contents, None, True)
self.all_contents[self.storage_index] = new_data
- self.my_uri.set_extension_params([self._k, self._segsize])
return None
# As actually implemented, MutableFilenode and MutableFileVersion
d = self.do_cli("put", "--mutable", "--mutable-type=mdmf", fn1)
def _got_cap((rc, out, err)):
self.failUnlessEqual(rc, 0)
- self.cap = out
+ self.cap = out.strip()
d.addCallback(_got_cap)
# Now try to write something to the cap using put.
data2 = "data2" * 100000
self.failUnlessEqual(rc, 0)
self.failUnlessEqual(out, data2)
d.addCallback(_got_data)
- # Now strip the extension information off of the cap and try
- # to put something to it.
- def _make_bare_cap(ignored):
- cap = self.cap.split(":")
- cap = ":".join(cap[:len(cap) - 2])
- self.cap = cap
- d.addCallback(_make_bare_cap)
+ # add some extension information to the cap and try to put something
+ # to it.
+ def _make_extended_cap(ignored):
+ self.cap = self.cap + ":Extension-Stuff"
+ d.addCallback(_make_extended_cap)
data3 = "data3" * 100000
fn3 = os.path.join(self.basedir, "data3")
fileutil.write(fn3, data3)
d = self.do_cli("put", "--mutable", "--mutable-type=sdmf", fn1)
def _got_cap((rc, out, err)):
self.failUnlessEqual(rc, 0)
- self.cap = out
+ self.cap = out.strip()
d.addCallback(_got_cap)
# Now try to write something to the cap using put.
data2 = "data2" * 100000
setup_py_uri = "URI:CHK:n7r3m6wmomelk4sep3kw5cvduq:os7ijw5c3maek7pg65e5254k2fzjflavtpejjyhshpsxuqzhcwwq:3:20:14861"
one_uri = "URI:LIT:n5xgk" # LIT for "one"
mut_write_uri = "URI:SSK:vfvcbdfbszyrsaxchgevhmmlii:euw4iw7bbnkrrwpzuburbhppuxhc3gwxv26f6imekhz7zyw2ojnq"
-mdmf_write_uri = "URI:MDMF:x533rhbm6kiehzl5kj3s44n5ie:4gif5rhneyd763ouo5qjrgnsoa3bg43xycy4robj2rf3tvmhdl3a:1:131072"
+mdmf_write_uri = "URI:MDMF:x533rhbm6kiehzl5kj3s44n5ie:4gif5rhneyd763ouo5qjrgnsoa3bg43xycy4robj2rf3tvmhdl3a"
empty_litdir_uri = "URI:DIR2-LIT:"
tiny_litdir_uri = "URI:DIR2-LIT:gqytunj2onug64tufqzdcosvkjetutcjkq5gw4tvm5vwszdgnz5hgyzufqydulbshj5x2lbm" # contains one child which is itself also LIT
mut_read_uri = "URI:SSK-RO:jf6wkflosyvntwxqcdo7a54jvm:euw4iw7bbnkrrwpzuburbhppuxhc3gwxv26f6imekhz7zyw2ojnq"
-mdmf_read_uri = "URI:MDMF-RO:d4cydxselputycfzkw6qgz4zv4:4gif5rhneyd763ouo5qjrgnsoa3bg43xycy4robj2rf3tvmhdl3a:1:131072"
+mdmf_read_uri = "URI:MDMF-RO:d4cydxselputycfzkw6qgz4zv4:4gif5rhneyd763ouo5qjrgnsoa3bg43xycy4robj2rf3tvmhdl3a"
future_write_uri = "x-tahoe-crazy://I_am_from_the_future."
future_read_uri = "x-tahoe-crazy-readonly://I_am_from_the_future."
future_nonascii_write_uri = u"x-tahoe-even-more-crazy://I_am_from_the_future_rw_\u263A".encode('utf-8')
return d
- def test_create_from_mdmf_writecap_with_extensions(self):
- # Test that the nodemaker is capable of creating an MDMF
- # filenode when given a writecap with extension parameters in
- # them.
- d = self.nodemaker.create_mutable_file(version=MDMF_VERSION)
- def _created(n):
- self.failUnless(isinstance(n, MutableFileNode))
- s = n.get_uri()
- # We need to cheat a little and delete the nodemaker's
- # cache, otherwise we'll get the same node instance back.
- self.failUnlessIn(":3:131073", s)
- n2 = self.nodemaker.create_from_cap(s)
-
- self.failUnlessEqual(n2.get_storage_index(), n.get_storage_index())
- self.failUnlessEqual(n.get_writekey(), n2.get_writekey())
- hints = n2._downloader_hints
- self.failUnlessEqual(hints['k'], 3)
- self.failUnlessEqual(hints['segsize'], 131073)
- d.addCallback(_created)
- return d
-
-
def test_create_from_mdmf_readcap(self):
d = self.nodemaker.create_mutable_file(version=MDMF_VERSION)
def _created(n):
return d
- def test_create_from_mdmf_readcap_with_extensions(self):
- # We should be able to create an MDMF filenode with the
- # extension parameters without it breaking.
- d = self.nodemaker.create_mutable_file(version=MDMF_VERSION)
- def _created(n):
- self.failUnless(isinstance(n, MutableFileNode))
- s = n.get_readonly_uri()
- self.failUnlessIn(":3:131073", s)
-
- n2 = self.nodemaker.create_from_cap(s)
- self.failUnless(isinstance(n2, MutableFileNode))
- self.failUnless(n2.is_readonly())
- self.failUnlessEqual(n.get_storage_index(), n2.get_storage_index())
- hints = n2._downloader_hints
- self.failUnlessEqual(hints["k"], 3)
- self.failUnlessEqual(hints["segsize"], 131073)
- d.addCallback(_created)
- return d
-
-
def test_internal_version_from_cap(self):
# MutableFileNodes and MutableFileVersions have an internal
# switch that tells them whether they're dealing with an SDMF or
d = self.nodemaker.create_mutable_file(version=MDMF_VERSION)
def _created(node):
self.uri = node.get_uri()
+ # also confirm that the cap has no extension fields
+ pieces = self.uri.split(":")
+ self.failUnlessEqual(len(pieces), 4)
return node.overwrite(MutableData("contents1" * 100000))
def _then(ignored):
return d
- def test_create_and_download_from_bare_mdmf_cap(self):
- # MDMF caps have extension parameters on them by default. We
- # need to make sure that they work without extension parameters.
- contents = MutableData("contents" * 100000)
- d = self.nodemaker.create_mutable_file(version=MDMF_VERSION,
- contents=contents)
- def _created(node):
- uri = node.get_uri()
- self._created = node
- self.failUnlessIn(":3:131073", uri)
- # Now strip that off the end of the uri, then try creating
- # and downloading the node again.
- bare_uri = uri.replace(":3:131073", "")
- assert ":3:131073" not in bare_uri
-
- return self.nodemaker.create_from_cap(bare_uri)
- d.addCallback(_created)
- def _created_bare(node):
- self.failUnlessEqual(node.get_writekey(),
- self._created.get_writekey())
- self.failUnlessEqual(node.get_readkey(),
- self._created.get_readkey())
- self.failUnlessEqual(node.get_storage_index(),
- self._created.get_storage_index())
- return node.download_best_version()
- d.addCallback(_created_bare)
- d.addCallback(lambda data:
- self.failUnlessEqual(data, "contents" * 100000))
- return d
-
-
def test_mdmf_write_count(self):
# Publishing an MDMF file should only cause one write for each
# share that is to be published. Otherwise, we introduce
return d
- def test_version_extension_api(self):
- # We need to define an API by which an uploader can set the
- # extension parameters, and by which a downloader can retrieve
- # extensions.
- d = self.do_upload_mdmf()
- d.addCallback(lambda ign: self.mdmf_node.get_best_mutable_version())
- def _got_version(version):
- hints = version.get_downloader_hints()
- # Should be empty at this point.
- self.failUnlessIn("k", hints)
- self.failUnlessEqual(hints['k'], 3)
- self.failUnlessIn('segsize', hints)
- self.failUnlessEqual(hints['segsize'], 131073)
- d.addCallback(_got_version)
- return d
-
-
- def test_extensions_from_cap(self):
- # If we initialize a mutable file with a cap that has extension
- # parameters in it and then grab the extension parameters using
- # our API, we should see that they're set correctly.
- d = self.do_upload_mdmf()
- def _then(ign):
- mdmf_uri = self.mdmf_node.get_uri()
- new_node = self.nm.create_from_cap(mdmf_uri)
- return new_node.get_best_mutable_version()
- d.addCallback(_then)
- def _got_version(version):
- hints = version.get_downloader_hints()
- self.failUnlessIn("k", hints)
- self.failUnlessEqual(hints["k"], 3)
- self.failUnlessIn("segsize", hints)
- self.failUnlessEqual(hints["segsize"], 131073)
- d.addCallback(_got_version)
- return d
-
-
- def test_extensions_from_upload(self):
- # If we create a new mutable file with some contents, we should
- # get back an MDMF cap with the right hints in place.
- contents = "foo bar baz" * 100000
- d = self.nm.create_mutable_file(contents, version=MDMF_VERSION)
- def _got_mutable_file(n):
- rw_uri = n.get_uri()
- expected_k = str(self.c.DEFAULT_ENCODING_PARAMETERS['k'])
- self.failUnlessIn(expected_k, rw_uri)
- # XXX: Get this more intelligently.
- self.failUnlessIn("131073", rw_uri)
-
- ro_uri = n.get_readonly_uri()
- self.failUnlessIn(expected_k, ro_uri)
- self.failUnlessIn("131073", ro_uri)
- d.addCallback(_got_mutable_file)
- return d
-
-
def test_cap_after_upload(self):
# If we create a new mutable file and upload things to it, and
# it's an MDMF file, we should get an MDMF cap back from that
-import re
+import os, re
from twisted.trial import unittest
from allmydata import uri
from allmydata.util import hashutil, base32
u4 = u2.get_verify_cap()
self.failUnlessReallyEqual(u4, u2)
- def test_mdmf_cap_extra_information(self):
- # MDMF caps can be arbitrarily extended after the fingerprint
- # and key/storage index fields.
+ def test_mdmf_cap_ignore_extensions(self):
+ # MDMF caps can be arbitrarily extended after the fingerprint and
+ # key/storage index fields. tahoe-1.9 is supposed to ignore any
+ # extensions, and not add any itself.
u1 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint)
- self.failUnlessEqual([], u1.get_extension_params())
-
- cap = u1.to_string()
- # Now let's append some fields. Say, 131073 (the segment size)
- # and 3 (the "k" encoding parameter).
- expected_extensions = []
- for e in ('131073', '3'):
- cap += (":%s" % e)
- expected_extensions.append(e)
-
- u2 = uri.WriteableMDMFFileURI.init_from_string(cap)
- self.failUnlessReallyEqual(self.writekey, u2.writekey)
- self.failUnlessReallyEqual(self.fingerprint, u2.fingerprint)
- self.failIf(u2.is_readonly())
- self.failUnless(u2.is_mutable())
-
- c2 = u2.to_string()
- u2n = uri.WriteableMDMFFileURI.init_from_string(c2)
- self.failUnlessReallyEqual(u2, u2n)
-
- # We should get the extra back when we ask for it.
- self.failUnlessEqual(expected_extensions, u2.get_extension_params())
-
- # These should be preserved through cap attenuation, too.
- u3 = u2.get_readonly()
- self.failUnlessReallyEqual(self.readkey, u3.readkey)
- self.failUnlessReallyEqual(self.fingerprint, u3.fingerprint)
- self.failUnless(u3.is_readonly())
- self.failUnless(u3.is_mutable())
- self.failUnlessEqual(expected_extensions, u3.get_extension_params())
-
- c3 = u3.to_string()
- u3n = uri.ReadonlyMDMFFileURI.init_from_string(c3)
- self.failUnlessReallyEqual(u3, u3n)
-
- u4 = u3.get_verify_cap()
- self.failUnlessReallyEqual(self.storage_index, u4.storage_index)
- self.failUnlessReallyEqual(self.fingerprint, u4.fingerprint)
- self.failUnless(u4.is_readonly())
- self.failIf(u4.is_mutable())
-
- c4 = u4.to_string()
- u4n = uri.MDMFVerifierURI.init_from_string(c4)
- self.failUnlessReallyEqual(u4n, u4)
-
- self.failUnlessEqual(expected_extensions, u4.get_extension_params())
-
-
- def test_sdmf_cap_extra_information(self):
- # For interface consistency, we define a method to get
- # extensions for SDMF files as well. This method must always
- # return no extensions, since SDMF files were not created with
- # extensions and cannot be modified to include extensions
- # without breaking older clients.
- u1 = uri.WriteableSSKFileURI(self.writekey, self.fingerprint)
cap = u1.to_string()
- u2 = uri.WriteableSSKFileURI.init_from_string(cap)
- self.failUnlessEqual([], u2.get_extension_params())
-
- def test_extension_character_range(self):
- # As written now, we shouldn't put things other than numbers in
- # the extension fields.
- writecap = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint).to_string()
- readcap = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint).to_string()
- vcap = uri.MDMFVerifierURI(self.storage_index, self.fingerprint).to_string()
- self.failUnlessRaises(uri.BadURIError,
- uri.WriteableMDMFFileURI.init_from_string,
- ("%s:invalid" % writecap))
- self.failUnlessRaises(uri.BadURIError,
- uri.ReadonlyMDMFFileURI.init_from_string,
- ("%s:invalid" % readcap))
- self.failUnlessRaises(uri.BadURIError,
- uri.MDMFVerifierURI.init_from_string,
- ("%s:invalid" % vcap))
+
+ cap2 = cap+":I COME FROM THE FUTURE"
+ u2 = uri.WriteableMDMFFileURI.init_from_string(cap2)
+ self.failUnlessReallyEqual(self.writekey, u2.writekey)
+ self.failUnlessReallyEqual(self.fingerprint, u2.fingerprint)
+ self.failIf(u2.is_readonly())
+ self.failUnless(u2.is_mutable())
+
+ cap3 = cap+":"+os.urandom(40) # parse *that*!
+ u3 = uri.WriteableMDMFFileURI.init_from_string(cap3)
+ self.failUnlessReallyEqual(self.writekey, u3.writekey)
+ self.failUnlessReallyEqual(self.fingerprint, u3.fingerprint)
+ self.failIf(u3.is_readonly())
+ self.failUnless(u3.is_mutable())
+
+ cap4 = u1.get_readonly().to_string()+":ooh scary future stuff"
+ u4 = uri.from_string_mutable_filenode(cap4)
+ self.failUnlessReallyEqual(self.readkey, u4.readkey)
+ self.failUnlessReallyEqual(self.fingerprint, u4.fingerprint)
+ self.failUnless(u4.is_readonly())
+ self.failUnless(u4.is_mutable())
+
+ cap5 = u1.get_verify_cap().to_string()+":spoilers!"
+ u5 = uri.from_string(cap5)
+ self.failUnlessReallyEqual(self.storage_index, u5.storage_index)
+ self.failUnlessReallyEqual(self.fingerprint, u5.fingerprint)
+ self.failUnless(u5.is_readonly())
+ self.failIf(u5.is_mutable())
def test_mdmf_valid_human_encoding(self):
# test that a valid cap (with and without the traditional
# separators) is recognized and accepted by the classes.
w1 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint)
- w2 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint,
- ['131073', '3'])
r1 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint)
- r2 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint,
- ['131073', '3'])
v1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint)
- v2 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint,
- ['131073', '3'])
- # These will yield six different caps.
- for o in (w1, w2, r1 , r2, v1, v2):
+ # These will yield three different caps.
+ for o in (w1, r1, v1):
url = base + o.to_string()
o1 = o.__class__.init_from_human_encoding(url)
self.failUnlessReallyEqual(o1, o)
- # Note that our cap will, by default, have : as separators.
+ # Note that our cap will, by default, have : as separators.
# But it's expected that users from, e.g., the WUI, will
# have %3A as a separator. We need to make sure that the
# initialization routine handles that, too.
# test that a valid cap (with and without the traditional
# separators) is recognized and accepted by the classes.
w1 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint)
- w2 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint,
- ['131073', '3'])
r1 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint)
- r2 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint,
- ['131073', '3'])
v1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint)
- v2 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint,
- ['131073', '3'])
- # These will yield six different caps.
- for o in (w1, w2, r1 , r2, v1, v2):
+ # These will yield three different caps.
+ for o in (w1, r1, v1):
url = base + o.to_string()
self.failUnlessRaises(uri.BadURIError,
o.__class__.init_from_human_encoding,
# test that a valid cap (with and without the traditional
# separators) is recognized and accepted by the classes.
w1 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint)
- w2 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint,
- ['131073', '3'])
r1 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint)
- r2 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint,
- ['131073', '3'])
v1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint)
- v2 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint,
- ['131073', '3'])
- # These will yield six different caps.
- for o in (w1, w2, r1 , r2, v1, v2):
+ # These will yield three different caps.
+ for o in (w1, r1, v1):
# not exhaustive, obviously...
url = base + o.to_string() + "foobarbaz"
url2 = base + "foobarbaz" + o.to_string()
u3 = uri.from_string_mutable_filenode(cap)
self.failUnlessEqual(u3, u1)
- # XXX: We should refactor the extension field into setUp
- u1 = uri.WriteableMDMFFileURI(self.writekey, self.fingerprint,
- ['131073', '3'])
- cap = u1.to_string()
- self.failUnless(uri.is_uri(cap))
- u2 = uri.from_string(cap)
- self.failUnlessReallyEqual(u1, u2)
- u3 = uri.from_string_mutable_filenode(cap)
- self.failUnlessEqual(u3, u1)
-
u1 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint)
cap = u1.to_string()
self.failUnless(uri.is_uri(cap))
u3 = uri.from_string_mutable_filenode(cap)
self.failUnlessEqual(u3, u1)
- u1 = uri.ReadonlyMDMFFileURI(self.readkey, self.fingerprint,
- ['131073', '3'])
- cap = u1.to_string()
- self.failUnless(uri.is_uri(cap))
- u2 = uri.from_string(cap)
- self.failUnlessReallyEqual(u1, u2)
- u3 = uri.from_string_mutable_filenode(cap)
- self.failUnlessEqual(u3, u1)
-
u1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint)
cap = u1.to_string()
self.failUnless(uri.is_uri(cap))
u3 = uri.from_string_verifier(cap)
self.failUnlessEqual(u3, u1)
- u1 = uri.MDMFVerifierURI(self.storage_index, self.fingerprint,
- ['131073', '3'])
- cap = u1.to_string()
- self.failUnless(uri.is_uri(cap))
- u2 = uri.from_string(cap)
- self.failUnlessReallyEqual(u1, u2)
- u3 = uri.from_string_verifier(cap)
- self.failUnlessEqual(u3, u1)
-
class Dirnode(testutil.ReallyEqualMixin, unittest.TestCase):
def test_pack(self):
d3 = uri.from_string(d2.to_string(), deep_immutable=True)
self.failUnlessIsInstance(d3, uri.UnknownURI)
- def test_mdmf_with_extensions(self):
- writekey = "\x01" * 16
- fingerprint = "\x02" * 32
- uri1 = uri.WriteableMDMFFileURI(writekey, fingerprint)
- d1 = uri.MDMFDirectoryURI(uri1)
- d1_uri = d1.to_string()
- # Add some extensions, verify that the URI is interpreted
- # correctly.
- d1_uri += ":3:131073"
- uri2 = uri.from_string(d1_uri)
- self.failUnlessIsInstance(uri2, uri.MDMFDirectoryURI)
- self.failUnless(IURI.providedBy(uri2))
- self.failUnless(IDirnodeURI.providedBy(uri2))
- self.failUnless(uri1.is_mutable())
- self.failIf(uri1.is_readonly())
-
- d2_uri = uri2.to_string()
- self.failUnlessIn(":3:131073", d2_uri)
-
- # Now attenuate, verify that the extensions persist
- ro_uri = uri2.get_readonly()
- self.failUnlessIsInstance(ro_uri, uri.ReadonlyMDMFDirectoryURI)
- self.failUnless(ro_uri.is_mutable())
- self.failUnless(ro_uri.is_readonly())
- self.failUnless(IURI.providedBy(ro_uri))
- self.failUnless(IDirnodeURI.providedBy(ro_uri))
- ro_uri_str = ro_uri.to_string()
- self.failUnlessIn(":3:131073", ro_uri_str)
-
def test_mdmf_attenuation(self):
writekey = "\x01" * 16
fingerprint = "\x02" * 32
return d
def test_GET_FILE_URI_mdmf_extensions(self):
- base = "/uri/%s" % urllib.quote("%s:3:131073" % self._quux_txt_uri)
- d = self.GET(base)
- d.addCallback(self.failUnlessIsQuuxDotTxt)
- return d
-
- def test_GET_FILE_URI_mdmf_bare_cap(self):
- cap_elements = self._quux_txt_uri.split(":")
- # 6 == expected cap length with two extensions.
- self.failUnlessEqual(len(cap_elements), 6)
-
- # Now lop off the extension parameters and stitch everything
- # back together
- quux_uri = ":".join(cap_elements[:len(cap_elements) - 2])
-
- # Now GET that. We should get back quux.
- base = "/uri/%s" % urllib.quote(quux_uri)
+ base = "/uri/%s" % urllib.quote("%s:RANDOMSTUFF" % self._quux_txt_uri)
d = self.GET(base)
d.addCallback(self.failUnlessIsQuuxDotTxt)
return d
return d
def test_PUT_FILE_URI_mdmf_extensions(self):
- base = "/uri/%s" % urllib.quote("%s:3:131073" % self._quux_txt_uri)
+ base = "/uri/%s" % urllib.quote("%s:EXTENSIONSTUFF" % self._quux_txt_uri)
self._quux_new_contents = "new_contents"
d = self.GET(base)
d.addCallback(lambda res: self.failUnlessIsQuuxDotTxt(res))
res))
return d
- def test_PUT_FILE_URI_mdmf_bare_cap(self):
- elements = self._quux_txt_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
-
- quux_uri = ":".join(elements[:len(elements) - 2])
- base = "/uri/%s" % urllib.quote(quux_uri)
- self._quux_new_contents = "new_contents" * 50000
-
- d = self.GET(base)
- d.addCallback(self.failUnlessIsQuuxDotTxt)
- d.addCallback(lambda ignored: self.PUT(base, self._quux_new_contents))
- d.addCallback(lambda ignored: self.GET(base))
- d.addCallback(lambda res:
- self.failUnlessEqual(res, self._quux_new_contents))
- return d
-
def test_PUT_FILE_URI_mdmf_readonly(self):
# We're not allowed to PUT things to a readonly cap.
base = "/uri/%s" % self._quux_txt_readonly_uri
return d
def test_GET_FILEURL_info_mdmf_extensions(self):
- d = self.GET("/uri/%s:3:131073?t=info" % self._quux_txt_uri)
+ d = self.GET("/uri/%s:STUFF?t=info" % self._quux_txt_uri)
def _got(res):
self.failUnlessIn("mutable file (mdmf)", res)
self.failUnlessIn(self._quux_txt_uri, res)
d.addCallback(_got)
return d
- def test_GET_FILEURL_info_mdmf_bare_cap(self):
- elements = self._quux_txt_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
-
- quux_uri = ":".join(elements[:len(elements) - 2])
- base = "/uri/%s?t=info" % urllib.quote(quux_uri)
- d = self.GET(base)
- def _got(res):
- self.failUnlessIn("mutable file (mdmf)", res)
- self.failUnlessIn(quux_uri, res)
- d.addCallback(_got)
- return d
-
def test_PUT_overwrite_only_files(self):
# create a directory, put a file in that directory.
contents, n, filecap = self.makefile(8)
d.addCallback(_got_json, "sdmf")
return d
- def test_GET_FILEURL_json_mdmf_extensions(self):
- # A GET invoked against a URL that includes an MDMF cap with
- # extensions should fetch the same JSON information as a GET
- # invoked against a bare cap.
- self._quux_txt_uri = "%s:3:131073" % self._quux_txt_uri
- self._quux_txt_readonly_uri = "%s:3:131073" % self._quux_txt_readonly_uri
- d = self.GET("/uri/%s?t=json" % urllib.quote(self._quux_txt_uri))
- d.addCallback(self.failUnlessIsQuuxJSON)
- return d
-
- def test_GET_FILEURL_json_mdmf_bare_cap(self):
- elements = self._quux_txt_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
-
- quux_uri = ":".join(elements[:len(elements) - 2])
- # so failUnlessIsQuuxJSON will work.
- self._quux_txt_uri = quux_uri
-
- # we need to alter the readonly URI in the same way, again so
- # failUnlessIsQuuxJSON will work
- elements = self._quux_txt_readonly_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
- quux_ro_uri = ":".join(elements[:len(elements) - 2])
- self._quux_txt_readonly_uri = quux_ro_uri
-
- base = "/uri/%s?t=json" % urllib.quote(quux_uri)
- d = self.GET(base)
- d.addCallback(self.failUnlessIsQuuxJSON)
- return d
-
- def test_GET_FILEURL_json_mdmf_bare_readonly_cap(self):
- elements = self._quux_txt_readonly_uri.split(":")
- self.failUnlessEqual(len(elements), 6)
-
- quux_readonly_uri = ":".join(elements[:len(elements) - 2])
- # so failUnlessIsQuuxJSON will work
- self._quux_txt_readonly_uri = quux_readonly_uri
- base = "/uri/%s?t=json" % quux_readonly_uri
- d = self.GET(base)
- # XXX: We may need to make a method that knows how to check for
- # readonly JSON, or else alter that one so that it knows how to
- # do that.
- d.addCallback(self.failUnlessIsQuuxJSON, readonly=True)
- return d
-
def test_GET_FILEURL_json_mdmf(self):
d = self.GET("/uri/%s?t=json" % urllib.quote(self._quux_txt_uri))
d.addCallback(self.failUnlessIsQuuxJSON)
SEP='(?::|%3A)'
NUMBER='([0-9]+)'
NUMBER_IGNORE='(?:[0-9]+)'
-OPTIONAL_EXTENSION_FIELD = '(' + SEP + '[0-9' + SEP + ']+|)'
# "human-encoded" URIs are allowed to come with a leading
# 'http://127.0.0.1:(8123|3456)/uri/' that will be ignored.
def get_verify_cap(self):
return SSKVerifierURI(self.storage_index, self.fingerprint)
- def get_extension_params(self):
- return []
-
- def set_extension_params(self, params):
- pass
-
class ReadonlySSKFileURI(_BaseURI):
implements(IURI, IMutableFileURI)
def get_verify_cap(self):
return SSKVerifierURI(self.storage_index, self.fingerprint)
- def get_extension_params(self):
- return []
-
- def set_extension_params(self, params):
- pass
-
class SSKVerifierURI(_BaseURI):
implements(IVerifierURI)
def get_verify_cap(self):
return self
- def get_extension_params(self):
- return []
-
- def set_extension_params(self, params):
- pass
-
class WriteableMDMFFileURI(_BaseURI):
implements(IURI, IMutableFileURI)
BASE_STRING='URI:MDMF:'
- STRING_RE=re.compile('^'+BASE_STRING+BASE32STR_128bits+':'+BASE32STR_256bits+OPTIONAL_EXTENSION_FIELD+'$')
- HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'MDMF'+SEP+BASE32STR_128bits+SEP+BASE32STR_256bits+OPTIONAL_EXTENSION_FIELD+'$')
+ STRING_RE=re.compile('^'+BASE_STRING+BASE32STR_128bits+':'+BASE32STR_256bits+'(:|$)')
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'MDMF'+SEP+BASE32STR_128bits+SEP+BASE32STR_256bits+'(:|$)')
- def __init__(self, writekey, fingerprint, params=[]):
+ def __init__(self, writekey, fingerprint):
self.writekey = writekey
self.readkey = hashutil.ssk_readkey_hash(writekey)
self.storage_index = hashutil.ssk_storage_index_hash(self.readkey)
assert len(self.storage_index) == 16
self.fingerprint = fingerprint
- self.extension = params
@classmethod
def init_from_human_encoding(cls, uri):
mo = cls.HUMAN_RE.search(uri)
if not mo:
raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
- params = filter(lambda x: x != '', re.split(SEP, mo.group(3)))
- return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)), params)
+ return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)))
@classmethod
def init_from_string(cls, uri):
mo = cls.STRING_RE.search(uri)
if not mo:
raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
- params = mo.group(3)
- params = filter(lambda x: x != '', params.split(":"))
- return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)), params)
+ return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)))
def to_string(self):
assert isinstance(self.writekey, str)
assert isinstance(self.fingerprint, str)
ret = 'URI:MDMF:%s:%s' % (base32.b2a(self.writekey),
base32.b2a(self.fingerprint))
- if self.extension:
- ret += ":"
- ret += ":".join(self.extension)
-
return ret
def __repr__(self):
return True
def get_readonly(self):
- return ReadonlyMDMFFileURI(self.readkey, self.fingerprint, self.extension)
+ return ReadonlyMDMFFileURI(self.readkey, self.fingerprint)
def get_verify_cap(self):
- return MDMFVerifierURI(self.storage_index, self.fingerprint, self.extension)
-
- def get_extension_params(self):
- return self.extension
-
- def set_extension_params(self, params):
- params = map(str, params)
- self.extension = params
+ return MDMFVerifierURI(self.storage_index, self.fingerprint)
class ReadonlyMDMFFileURI(_BaseURI):
implements(IURI, IMutableFileURI)
BASE_STRING='URI:MDMF-RO:'
- STRING_RE=re.compile('^' +BASE_STRING+BASE32STR_128bits+':'+BASE32STR_256bits+OPTIONAL_EXTENSION_FIELD+'$')
- HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'MDMF-RO'+SEP+BASE32STR_128bits+SEP+BASE32STR_256bits+OPTIONAL_EXTENSION_FIELD+'$')
+ STRING_RE=re.compile('^' +BASE_STRING+BASE32STR_128bits+':'+BASE32STR_256bits+'(:|$)')
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'MDMF-RO'+SEP+BASE32STR_128bits+SEP+BASE32STR_256bits+'(:|$)')
- def __init__(self, readkey, fingerprint, params=[]):
+ def __init__(self, readkey, fingerprint):
self.readkey = readkey
self.storage_index = hashutil.ssk_storage_index_hash(self.readkey)
assert len(self.storage_index) == 16
self.fingerprint = fingerprint
- self.extension = params
@classmethod
def init_from_human_encoding(cls, uri):
mo = cls.HUMAN_RE.search(uri)
if not mo:
raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
- params = mo.group(3)
- params = filter(lambda x: x!= '', re.split(SEP, params))
- return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)), params)
+ return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)))
@classmethod
def init_from_string(cls, uri):
if not mo:
raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
- params = mo.group(3)
- params = filter(lambda x: x != '', params.split(":"))
- return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)), params)
+ return cls(base32.a2b(mo.group(1)), base32.a2b(mo.group(2)))
def to_string(self):
assert isinstance(self.readkey, str)
assert isinstance(self.fingerprint, str)
ret = 'URI:MDMF-RO:%s:%s' % (base32.b2a(self.readkey),
base32.b2a(self.fingerprint))
- if self.extension:
- ret += ":"
- ret += ":".join(self.extension)
-
return ret
def __repr__(self):
return self
def get_verify_cap(self):
- return MDMFVerifierURI(self.storage_index, self.fingerprint, self.extension)
-
- def get_extension_params(self):
- return self.extension
-
- def set_extension_params(self, params):
- params = map(str, params)
- self.extension = params
+ return MDMFVerifierURI(self.storage_index, self.fingerprint)
class MDMFVerifierURI(_BaseURI):
implements(IVerifierURI)
BASE_STRING='URI:MDMF-Verifier:'
- STRING_RE=re.compile('^'+BASE_STRING+BASE32STR_128bits+':'+BASE32STR_256bits+OPTIONAL_EXTENSION_FIELD+'$')
- HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'MDMF-Verifier'+SEP+BASE32STR_128bits+SEP+BASE32STR_256bits+OPTIONAL_EXTENSION_FIELD+'$')
+ STRING_RE=re.compile('^'+BASE_STRING+BASE32STR_128bits+':'+BASE32STR_256bits+'(:|$)')
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'MDMF-Verifier'+SEP+BASE32STR_128bits+SEP+BASE32STR_256bits+'(:|$)')
- def __init__(self, storage_index, fingerprint, params=[]):
+ def __init__(self, storage_index, fingerprint):
assert len(storage_index) == 16
self.storage_index = storage_index
self.fingerprint = fingerprint
- self.extension = params
@classmethod
def init_from_human_encoding(cls, uri):
mo = cls.HUMAN_RE.search(uri)
if not mo:
raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
- params = mo.group(3)
- params = filter(lambda x: x != '', re.split(SEP, params))
- return cls(si_a2b(mo.group(1)), base32.a2b(mo.group(2)), params)
+ return cls(si_a2b(mo.group(1)), base32.a2b(mo.group(2)))
@classmethod
def init_from_string(cls, uri):
mo = cls.STRING_RE.search(uri)
if not mo:
raise BadURIError("'%s' doesn't look like a %s cap" % (uri, cls))
- params = mo.group(3)
- params = filter(lambda x: x != '', params.split(":"))
- return cls(si_a2b(mo.group(1)), base32.a2b(mo.group(2)), params)
+ return cls(si_a2b(mo.group(1)), base32.a2b(mo.group(2)))
def to_string(self):
assert isinstance(self.storage_index, str)
assert isinstance(self.fingerprint, str)
ret = 'URI:MDMF-Verifier:%s:%s' % (si_b2a(self.storage_index),
base32.b2a(self.fingerprint))
- if self.extension:
- ret += ':'
- ret += ":".join(self.extension)
-
return ret
def is_readonly(self):
def get_verify_cap(self):
return self
- def get_extension_params(self):
- return self.extension
-
class _DirectoryBaseURI(_BaseURI):
implements(IURI, IDirnodeURI)
def __init__(self, filenode_uri=None):
projects / tahoe-lafs / tahoe-lafs.git / commitdiff