src/allmydata/uri.py

   1
   2 import re
   3 from allmydata.util import idlib, hashutil
   4
   5 def get_uri_type(uri):
   6     assert uri.startswith("URI:")
   7     if uri.startswith("URI:DIR:"):
   8         return "DIR"
   9     if uri.startswith("URI:DIR-RO:"):
  10         return "DIR-RO"
  11     if uri.startswith("URI:LIT:"):
  12         return "LIT"
  13     return "CHK"
  14
  15 # the URI shall be an ascii representation of the file. It shall contain
  16 # enough information to retrieve and validate the contents. It shall be
  17 # expressed in a limited character set (namely [TODO]).
  18
  19 def pack_uri(storage_index, key, uri_extension_hash,
  20              needed_shares, total_shares, size):
  21     # applications should pass keyword parameters into this
  22     assert isinstance(storage_index, str)
  23     assert len(storage_index) == 32 # sha256 hash
  24
  25     assert isinstance(uri_extension_hash, str)
  26     assert len(uri_extension_hash) == 32 # sha56 hash
  27
  28     assert isinstance(key, str)
  29     assert len(key) == 16 # AES-128
  30     assert isinstance(needed_shares, int)
  31     assert isinstance(total_shares, int)
  32     assert isinstance(size, (int,long))
  33
  34     return "URI:%s:%s:%s:%d:%d:%d" % (idlib.b2a(storage_index), idlib.b2a(key),
  35                                       idlib.b2a(uri_extension_hash),
  36                                       needed_shares, total_shares, size)
  37
  38
  39 def unpack_uri(uri):
  40     assert uri.startswith("URI:"), uri
  41     d = {}
  42     (header,
  43      storage_index_s, key_s, uri_extension_hash_s,
  44      needed_shares_s, total_shares_s, size_s) = uri.split(":")
  45     assert header == "URI"
  46     d['storage_index'] = idlib.a2b(storage_index_s)
  47     d['key'] = idlib.a2b(key_s)
  48     d['uri_extension_hash'] = idlib.a2b(uri_extension_hash_s)
  49     d['needed_shares'] = int(needed_shares_s)
  50     d['total_shares'] = int(total_shares_s)
  51     d['size'] = int(size_s)
  52     return d
  53
  54
  55 def pack_extension(data):
  56     pieces = []
  57     for k in sorted(data.keys()):
  58         value = data[k]
  59         if isinstance(value, (int, long)):
  60             value = "%d" % value
  61         assert isinstance(value, str), k
  62         assert re.match(r'^[a-zA-Z_\-]+$', k)
  63         pieces.append(k + ":" + hashutil.netstring(value))
  64     uri_extension = "".join(pieces)
  65     return uri_extension
  66
  67 def unpack_extension(data):
  68     d = {}
  69     while data:
  70         colon = data.index(":")
  71         key = data[:colon]
  72         data = data[colon+1:]
  73
  74         colon = data.index(":")
  75         number = data[:colon]
  76         length = int(number)
  77         data = data[colon+1:]
  78
  79         value = data[:length]
  80         assert data[length] == ","
  81         data = data[length+1:]
  82
  83         d[key] = value
  84
  85     # convert certain things to numbers
  86     for intkey in ("size", "segment_size", "num_segments",
  87                    "needed_shares", "total_shares"):
  88         if intkey in d:
  89             d[intkey] = int(d[intkey])
  90     return d
  91
  92
  93 def unpack_extension_readable(data):
  94     unpacked = unpack_extension(data)
  95     for k in sorted(unpacked.keys()):
  96         if "hash" in k:
  97             unpacked[k] = idlib.b2a(unpacked[k])
  98     return unpacked
  99
 100 def pack_lit(data):
 101     return "URI:LIT:%s" % idlib.b2a(data)
 102
 103 def unpack_lit(uri):
 104     assert uri.startswith("URI:LIT:")
 105     data_s = uri[len("URI:LIT:"):]
 106     return idlib.a2b(data_s)
 107
 108
 109 def is_dirnode_uri(uri):
 110     return uri.startswith("URI:DIR:") or uri.startswith("URI:DIR-RO:")
 111 def is_mutable_dirnode_uri(uri):
 112     return uri.startswith("URI:DIR:")
 113 def unpack_dirnode_uri(uri):
 114     assert is_dirnode_uri(uri)
 115     # URI:DIR:furl:key
 116     #  but note that the furl contains colons
 117     for prefix in ("URI:DIR:", "URI:DIR-RO:"):
 118         if uri.startswith(prefix):
 119             uri = uri[len(prefix):]
 120             break
 121     else:
 122         assert 0
 123     colon = uri.rindex(":")
 124     furl = uri[:colon]
 125     key = uri[colon+1:]
 126     return furl, idlib.a2b(key)
 127
 128 def make_immutable_dirnode_uri(mutable_uri):
 129     assert is_mutable_dirnode_uri(mutable_uri)
 130     furl, writekey = unpack_dirnode_uri(mutable_uri)
 131     readkey = hashutil.dir_read_key_hash(writekey)
 132     return "URI:DIR-RO:%s:%s" % (furl, idlib.b2a(readkey))
 133
 134 def pack_dirnode_uri(furl, writekey):
 135     return "URI:DIR:%s:%s" % (furl, idlib.b2a(writekey))