-import re
+import re, urllib
from zope.interface import implements
from twisted.python.components import registerAdapter
from allmydata.util import idlib, hashutil
# enough information to retrieve and validate the contents. It shall be
# expressed in a limited character set (namely [TODO]).
-ZBASE32CHAR = "[ybndrfg8ejkmcpqxot1uwisza345h769]" # excludes l, 0, 2, and v
-ZBASE32CHAR_3bits = "[yoearcwh]"
-ZBASE32CHAR_1bits = "[yo]"
-ZBASE32STR_128bits = "%s{25}%s" % (ZBASE32CHAR, ZBASE32CHAR_3bits)
-ZBASE32STR_256bits = "%s{51}%s" % (ZBASE32CHAR, ZBASE32CHAR_1bits)
-COLON="(:|%3A)"
+ZBASE32STR_128bits = '(%s{25}%s)' % (idlib.ZBASE32CHAR, idlib.ZBASE32CHAR_3bits)
+ZBASE32STR_256bits = '(%s{51}%s)' % (idlib.ZBASE32CHAR, idlib.ZBASE32CHAR_1bits)
-# Writeable SSK bits
-WSSKBITS= "(%s)%s(%s)" % (ZBASE32STR_128bits, COLON, ZBASE32STR_256bits)
+SEP='(?::|%3A)'
+NUMBER='([0-9]+)'
# URIs (soon to be renamed "caps") are always allowed to come with a leading
-# "http://127.0.0.1:8123/uri/" that will be ignored.
-OPTIONALHTTPLEAD=r'(https?://(127.0.0.1|localhost):8123/uri/)?'
-
-# Writeable SSK URI
-WriteableSSKFileURI_RE=re.compile("^%sURI%sSSK%s%s$" % (OPTIONALHTTPLEAD, COLON, COLON, WSSKBITS))
-
-# NewDirectory Read-Write URI
-DirnodeURI_RE=re.compile("^%sURI%sDIR2%s%s/?$" % (OPTIONALHTTPLEAD, COLON, COLON, WSSKBITS))
+# 'http://127.0.0.1:8123/uri/' that will be ignored.
+OPTIONALHTTPLEAD=r'(?:https?://(127.0.0.1|localhost):8123/uri/)?'
class _BaseURI:
if cmp(self.__class__, them.__class__):
return cmp(self.__class__, them.__class__)
return cmp(self.to_string(), them.to_string())
+ def to_human_encoding(self):
+ return 'http://127.0.0.1:8123/uri/'+self.to_string()
class CHKFileURI(_BaseURI):
implements(IURI, IFileURI)
- def __init__(self, **kwargs):
- # construct me with kwargs, since there are so many of them
- if not kwargs:
- return
- keys = ("key", "uri_extension_hash",
- "needed_shares", "total_shares", "size")
- for name in kwargs:
- if name in keys:
- value = kwargs[name]
- setattr(self, name, value)
- else:
- raise TypeError("CHKFileURI does not accept '%s=' argument"
- % name)
+ STRING_RE=re.compile('^URI:CHK:'+ZBASE32STR_128bits+':'+
+ ZBASE32STR_256bits+':'+NUMBER+':'+NUMBER+':'+NUMBER+
+ '$')
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'CHK'+SEP+
+ ZBASE32STR_128bits+SEP+ZBASE32STR_256bits+SEP+NUMBER+
+ SEP+NUMBER+SEP+NUMBER+'$')
+
+ def __init__(self, key, uri_extension_hash, needed_shares, total_shares,
+ size):
+ self.key = key
+ self.uri_extension_hash = uri_extension_hash
+ self.needed_shares = needed_shares
+ self.total_shares = total_shares
+ self.size = size
self.storage_index = hashutil.storage_index_chk_hash(self.key)
+ assert len(self.storage_index) == 16
+ self.storage_index = hashutil.storage_index_chk_hash(key)
+ assert len(self.storage_index) == 16 # sha256 hash truncated to 128
+
+ @classmethod
+ def init_from_human_encoding(cls, uri):
+ mo = cls.HUMAN_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)),
+ int(mo.group(3)), int(mo.group(4)), int(mo.group(5)))
@classmethod
def init_from_string(cls, uri):
- assert uri.startswith("URI:CHK:"), uri
- d = {}
- (header_uri, header_chk,
- key_s, uri_extension_hash_s,
- needed_shares_s, total_shares_s, size_s) = uri.split(":")
- assert header_uri == "URI"
- assert header_chk == "CHK"
-
- key = idlib.a2b(key_s)
- assert isinstance(key, str)
- assert len(key) == 16 # AES-128
-
- storage_index = hashutil.storage_index_chk_hash(key)
- assert isinstance(storage_index, str)
- assert len(storage_index) == 16 # sha256 hash truncated to 128
-
- uri_extension_hash = idlib.a2b(uri_extension_hash_s)
- assert isinstance(uri_extension_hash, str)
- assert len(uri_extension_hash) == 32 # sha256 hash
-
- needed_shares = int(needed_shares_s)
- total_shares = int(total_shares_s)
- size = int(size_s)
- return cls(key=key, uri_extension_hash=uri_extension_hash,
- needed_shares=needed_shares, total_shares=total_shares,
- size=size)
+ mo = cls.STRING_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)),
+ int(mo.group(3)), int(mo.group(4)), int(mo.group(5)))
def to_string(self):
assert isinstance(self.needed_shares, int)
assert isinstance(self.total_shares, int)
assert isinstance(self.size, (int,long))
- return ("URI:CHK:%s:%s:%d:%d:%d" %
+ return ('URI:CHK:%s:%s:%d:%d:%d' %
(idlib.b2a(self.key),
idlib.b2a(self.uri_extension_hash),
self.needed_shares,
class CHKFileVerifierURI(_BaseURI):
implements(IVerifierURI)
- def __init__(self, **kwargs):
- # construct me with kwargs, since there are so many of them
- if not kwargs:
- return
- self.populate(**kwargs)
+ STRING_RE=re.compile('^URI:CHK-Verifier:'+ZBASE32STR_128bits+':'+
+ ZBASE32STR_256bits+':'+NUMBER+':'+NUMBER+':'+NUMBER)
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'CHK-Verifier'+SEP+
+ ZBASE32STR_128bits+SEP+ZBASE32STR_256bits+SEP+NUMBER+
+ SEP+NUMBER+SEP+NUMBER)
- def populate(self, storage_index, uri_extension_hash,
+ def __init__(self, storage_index, uri_extension_hash,
needed_shares, total_shares, size):
+ assert len(storage_index) == 16
self.storage_index = storage_index
self.uri_extension_hash = uri_extension_hash
self.needed_shares = needed_shares
self.total_shares = total_shares
self.size = size
+ @classmethod
+ def init_from_human_encoding(cls, uri):
+ mo = cls.HUMAN_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)),
+ int(mo.group(3)), int(mo.group(4)), int(mo.group(5)))
+
@classmethod
def init_from_string(cls, uri):
- assert uri.startswith("URI:CHK-Verifier:"), uri
- d = {}
- (header_uri, header_chk,
- storage_index_s, uri_extension_hash_s,
- needed_shares_s, total_shares_s, size_s) = uri.split(":")
- assert header_uri == "URI"
- assert header_chk == "CHK-Verifier"
-
- storage_index = idlib.a2b(storage_index_s)
- assert isinstance(storage_index, str)
- assert len(storage_index) == 16 # sha256 hash truncated to 128
-
- uri_extension_hash = idlib.a2b(uri_extension_hash_s)
- assert isinstance(uri_extension_hash, str)
- assert len(uri_extension_hash) == 32 # sha256 hash
-
- needed_shares = int(needed_shares_s)
- total_shares = int(total_shares_s)
- size = int(size_s)
-
- return cls(storage_index=storage_index,
- uri_extension_hash=uri_extension_hash,
- needed_shares=needed_shares,
- total_shares=total_shares,
- size=size)
+ mo = cls.STRING_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)),
+ int(mo.group(3)), int(mo.group(4)), int(mo.group(5)))
def to_string(self):
assert isinstance(self.needed_shares, int)
assert isinstance(self.total_shares, int)
assert isinstance(self.size, (int,long))
- return ("URI:CHK-Verifier:%s:%s:%d:%d:%d" %
+ return ('URI:CHK-Verifier:%s:%s:%d:%d:%d' %
(idlib.b2a(self.storage_index),
idlib.b2a(self.uri_extension_hash),
self.needed_shares,
class LiteralFileURI(_BaseURI):
implements(IURI, IFileURI)
+ STRING_RE=re.compile('^URI:LIT:'+idlib.ZBASE32STR_anybytes+'$')
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'LIT'+SEP+idlib.ZBASE32STR_anybytes+'$')
+
def __init__(self, data=None):
if data is not None:
self.data = data
+ @classmethod
+ def init_from_human_encoding(cls, uri):
+ mo = cls.HUMAN_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)))
+
@classmethod
def init_from_string(cls, uri):
- assert uri.startswith("URI:LIT:")
- data_s = uri[len("URI:LIT:"):]
- return cls(idlib.a2b(data_s))
+ mo = cls.STRING_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)))
def to_string(self):
- return "URI:LIT:%s" % idlib.b2a(self.data)
+ return 'URI:LIT:%s' % idlib.b2a(self.data)
def is_readonly(self):
return True
class WriteableSSKFileURI(_BaseURI):
implements(IURI, IMutableFileURI)
+ BASE_STRING='URI:SSK:'
+ STRING_RE=re.compile('^'+BASE_STRING+ZBASE32STR_128bits+':'+
+ ZBASE32STR_256bits+'$')
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'SSK'+SEP+
+ ZBASE32STR_128bits+SEP+ZBASE32STR_256bits+'$')
+
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
@classmethod
def init_from_human_encoding(cls, uri):
- mo = WriteableSSKFileURI_RE.search(uri)
- assert mo
- return cls(idlib.a2b(mo.group(5)), idlib.a2b(mo.group(7)))
+ mo = cls.HUMAN_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)))
@classmethod
def init_from_string(cls, uri):
- assert uri.startswith("URI:SSK:"), uri
- (header_uri, header_ssk, writekey_s, fingerprint_s) = uri.split(":")
- return cls(idlib.a2b(writekey_s), idlib.a2b(fingerprint_s))
-
- def to_human_encoding(self):
- assert isinstance(self.writekey, str)
- assert isinstance(self.fingerprint, str)
- return "http://127.0.0.1:8123/uri/URI:SSK:%s:%s" % (idlib.b2a(self.writekey), idlib.b2a(self.fingerprint))
+ mo = cls.STRING_RE.search(uri)
+ assert mo, (uri, cls)
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)))
def to_string(self):
assert isinstance(self.writekey, str)
assert isinstance(self.fingerprint, str)
- return "URI:SSK:%s:%s" % (idlib.b2a(self.writekey),
+ return 'URI:SSK:%s:%s' % (idlib.b2a(self.writekey),
idlib.b2a(self.fingerprint))
def __repr__(self):
class ReadonlySSKFileURI(_BaseURI):
implements(IURI, IMutableFileURI)
+ BASE_STRING='URI:SSK-RO:'
+ STRING_RE=re.compile('^URI:SSK-RO:'+ZBASE32STR_128bits+':'+ZBASE32STR_256bits+'$')
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'SSK-RO'+SEP+ZBASE32STR_128bits+SEP+ZBASE32STR_256bits+'$')
+
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
+ @classmethod
+ def init_from_human_encoding(cls, uri):
+ mo = cls.HUMAN_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)))
+
@classmethod
def init_from_string(cls, uri):
- assert uri.startswith("URI:SSK-RO:"), uri
- (header_uri, header_ssk, readkey_s, fingerprint_s) = uri.split(":")
- return cls(idlib.a2b(readkey_s), idlib.a2b(fingerprint_s))
+ mo = cls.STRING_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)))
def to_string(self):
assert isinstance(self.readkey, str)
assert isinstance(self.fingerprint, str)
- return "URI:SSK-RO:%s:%s" % (idlib.b2a(self.readkey),
+ return 'URI:SSK-RO:%s:%s' % (idlib.b2a(self.readkey),
idlib.b2a(self.fingerprint))
def __repr__(self):
class SSKVerifierURI(_BaseURI):
implements(IVerifierURI)
+ BASE_STRING='URI:SSK-Verifier:'
+ STRING_RE=re.compile('^'+BASE_STRING+ZBASE32STR_128bits+':'+ZBASE32STR_256bits+'$')
+ HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'SSK-RO'+SEP+ZBASE32STR_128bits+SEP+ZBASE32STR_256bits+'$')
+
def __init__(self, storage_index, fingerprint):
+ assert len(storage_index) == 16
self.storage_index = storage_index
self.fingerprint = fingerprint
+ @classmethod
+ def init_from_human_encoding(cls, uri):
+ mo = cls.HUMAN_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)))
+
@classmethod
def init_from_string(cls, uri):
- assert uri.startswith("URI:SSK-Verifier:"), uri
- (header_uri, header_ssk,
- storage_index_s, fingerprint_s) = uri.split(":")
- return cls(idlib.a2b(storage_index_s), idlib.a2b(fingerprint_s))
+ mo = cls.STRING_RE.search(uri)
+ assert mo, uri
+ return cls(idlib.a2b(mo.group(1)), idlib.a2b(mo.group(2)))
def to_string(self):
assert isinstance(self.storage_index, str)
assert isinstance(self.fingerprint, str)
- return "URI:SSK-Verifier:%s:%s" % (idlib.b2a(self.storage_index),
+ return 'URI:SSK-Verifier:%s:%s' % (idlib.b2a(self.storage_index),
idlib.b2a(self.fingerprint))
class _NewDirectoryBaseURI(_BaseURI):
def __repr__(self):
return "<%s %s>" % (self.__class__.__name__, self.abbrev())
+ @classmethod
+ def init_from_string(cls, uri):
+ mo = cls.BASE_STRING_RE.search(uri)
+ assert mo, (uri, cls)
+ bits = uri[mo.end():]
+ fn = cls.INNER_URI_CLASS.init_from_string(
+ cls.INNER_URI_CLASS.BASE_STRING+bits)
+ return cls(fn)
+
+ @classmethod
+ def init_from_human_encoding(cls, uri):
+ mo = cls.BASE_HUMAN_RE.search(uri)
+ assert mo, uri
+ bits = uri[mo.end():]
+ while bits and bits[-1] == '/':
+ bits = bits[:-1]
+ fn = cls.INNER_URI_CLASS.init_from_string(
+ cls.INNER_URI_CLASS.BASE_STRING+urllib.unquote(bits))
+ return cls(fn)
+
+ def to_string(self):
+ fnuri = self._filenode_uri.to_string()
+ mo = re.match(self.INNER_URI_CLASS.BASE_STRING, fnuri)
+ assert mo, fnuri
+ bits = fnuri[mo.end():]
+ return self.BASE_STRING+bits
+
def abbrev(self):
return self._filenode_uri.to_string().split(':')[2][:5]
class NewDirectoryURI(_NewDirectoryBaseURI):
implements(INewDirectoryURI)
+
+ BASE_STRING='URI:DIR2:'
+ BASE_STRING_RE=re.compile('^'+BASE_STRING)
+ BASE_HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'DIR2'+SEP)
+ INNER_URI_CLASS=WriteableSSKFileURI
+
def __init__(self, filenode_uri=None):
if filenode_uri:
assert not filenode_uri.is_readonly()
_NewDirectoryBaseURI.__init__(self, filenode_uri)
- @classmethod
- def init_from_string(cls, uri):
- assert uri.startswith("URI:DIR2:")
- (header_uri, header_dir2, bits) = uri.split(":", 2)
- fn = WriteableSSKFileURI.init_from_string("URI:SSK:" + bits)
- return cls(fn)
-
- def to_string(self):
- assert isinstance(self._filenode_uri, WriteableSSKFileURI)
- fn_u = self._filenode_uri.to_string()
- (header_uri, header_ssk, bits) = fn_u.split(":", 2)
- return "URI:DIR2:" + bits
-
def is_readonly(self):
return False
class ReadonlyNewDirectoryURI(_NewDirectoryBaseURI):
implements(IReadonlyNewDirectoryURI)
+
+ BASE_STRING='URI:DIR2-RO:'
+ BASE_STRING_RE=re.compile('^'+BASE_STRING)
+ BASE_HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'DIR2-RO'+SEP)
+ INNER_URI_CLASS=ReadonlySSKFileURI
+
def __init__(self, filenode_uri=None):
if filenode_uri:
assert filenode_uri.is_readonly()
_NewDirectoryBaseURI.__init__(self, filenode_uri)
- @classmethod
- def init_from_string(cls, uri):
- assert uri.startswith("URI:DIR2-RO:")
- (header_uri, header_dir2, bits) = uri.split(":", 2)
- fn = ReadonlySSKFileURI.init_from_string("URI:SSK-RO:" + bits)
- return cls(fn)
-
- def to_string(self):
- assert isinstance(self._filenode_uri, ReadonlySSKFileURI)
- fn_u = self._filenode_uri.to_string()
- (header_uri, header_ssk, bits) = fn_u.split(":", 2)
- return "URI:DIR2-RO:" + bits
-
def is_readonly(self):
return True
def get_readonly(self):
return self
-class NewDirectoryURIVerifier(_BaseURI):
+class NewDirectoryURIVerifier(_NewDirectoryBaseURI):
implements(IVerifierURI)
+ BASE_STRING='URI:DIR2-Verifier:'
+ BASE_STRING_RE=re.compile('^'+BASE_STRING)
+ BASE_HUMAN_RE=re.compile('^'+OPTIONALHTTPLEAD+'URI'+SEP+'DIR2-Verifier'+SEP)
+ INNER_URI_CLASS=SSKVerifierURI
+
def __init__(self, filenode_uri=None):
if filenode_uri:
filenode_uri = IVerifierURI(filenode_uri)
self._filenode_uri = filenode_uri
- @classmethod
- def init_from_string(cls, uri):
- assert uri.startswith("URI:DIR2-Verifier:")
- (header_uri, header_dir2, bits) = uri.split(":", 2)
- fv = SSKVerifierURI.init_from_string("URI:SSK-Verifier:" + bits)
- return cls(fv)
-
- def to_string(self):
- assert isinstance(self._filenode_uri, SSKVerifierURI)
- fn_u = self._filenode_uri.to_string()
- (header_uri, header_ssk, bits) = fn_u.split(":", 2)
- return "URI:DIR2-Verifier:" + bits
-
def get_filenode_uri(self):
return self._filenode_uri
def from_string(s):
- if s.startswith("URI:CHK:"):
+ if s.startswith('URI:CHK:'):
return CHKFileURI.init_from_string(s)
- elif s.startswith("URI:CHK-Verifier:"):
+ elif s.startswith('URI:CHK-Verifier:'):
return CHKFileVerifierURI.init_from_string(s)
- elif s.startswith("URI:LIT:"):
+ elif s.startswith('URI:LIT:'):
return LiteralFileURI.init_from_string(s)
- elif s.startswith("URI:SSK:"):
+ elif s.startswith('URI:SSK:'):
return WriteableSSKFileURI.init_from_string(s)
- elif s.startswith("URI:SSK-RO:"):
+ elif s.startswith('URI:SSK-RO:'):
return ReadonlySSKFileURI.init_from_string(s)
- elif s.startswith("URI:SSK-Verifier:"):
+ elif s.startswith('URI:SSK-Verifier:'):
return SSKVerifierURI.init_from_string(s)
- elif s.startswith("URI:DIR2:"):
+ elif s.startswith('URI:DIR2:'):
return NewDirectoryURI.init_from_string(s)
- elif s.startswith("URI:DIR2-RO:"):
+ elif s.startswith('URI:DIR2-RO:'):
return ReadonlyNewDirectoryURI.init_from_string(s)
- elif s.startswith("URI:DIR2-Verifier:"):
+ elif s.startswith('URI:DIR2-Verifier:'):
return NewDirectoryURIVerifier.init_from_string(s)
else:
raise TypeError("unknown URI type: %s.." % s[:12])
registerAdapter(from_string_dirnode, str, IDirnodeURI)
-def is_string_newdirnode_rw(s):
- return DirnodeURI_RE.search(s)
-
def from_string_filenode(s):
u = from_string(s)
assert IFileURI.providedBy(u)
value = "%d" % value
assert isinstance(value, str), k
assert re.match(r'^[a-zA-Z_\-]+$', k)
- pieces.append(k + ":" + hashutil.netstring(value))
- uri_extension = "".join(pieces)
+ pieces.append(k + ':' + hashutil.netstring(value))
+ uri_extension = ''.join(pieces)
return uri_extension
def unpack_extension(data):
d = {}
while data:
- colon = data.index(":")
+ colon = data.index(':')
key = data[:colon]
data = data[colon+1:]
- colon = data.index(":")
+ colon = data.index(':')
number = data[:colon]
length = int(number)
data = data[colon+1:]
value = data[:length]
- assert data[length] == ","
+ assert data[length] == ','
data = data[length+1:]
d[key] = value
# convert certain things to numbers
- for intkey in ("size", "segment_size", "num_segments",
- "needed_shares", "total_shares"):
+ for intkey in ('size', 'segment_size', 'num_segments',
+ 'needed_shares', 'total_shares'):
if intkey in d:
d[intkey] = int(d[intkey])
return d
def unpack_extension_readable(data):
unpacked = unpack_extension(data)
for k in sorted(unpacked.keys()):
- if "hash" in k:
+ if 'hash' in k:
unpacked[k] = idlib.b2a(unpacked[k])
return unpacked
projects / tahoe-lafs / tahoe-lafs.git / commitdiff