mirror of
https://github.com/tahoe-lafs/tahoe-lafs.git
synced 2025-02-01 00:45:52 +00:00
make more precise regexp for WriteableSSKFileURI and DirnodeURI and use it in unit tests
Also allow an optional leading "http://127.0.0.1:8123/uri/". Also fix a few unit tests to generate bogus Dirnode URIs of the modern form instead of the former form.
This commit is contained in:
parent
ea43753386
commit
9848d2043d
@ -13,7 +13,31 @@ Nodeid = StringConstraint(maxLength=20,
|
||||
FURL = StringConstraint(1000)
|
||||
StorageIndex = StringConstraint(16)
|
||||
URI = StringConstraint(300) # kind of arbitrary
|
||||
DirnodeURI = StringConstraint(300, regexp=r'^URI:DIR(-RO)?:pb://[a-z0-9]+@[^/]+/[^:]+:[a-z0-9]+$')
|
||||
|
||||
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)"
|
||||
|
||||
# Writeable SSK bits
|
||||
WSSKBITS= "%s%s%s" % (ZBASE32STR_128bits, COLON, ZBASE32STR_256bits)
|
||||
|
||||
# 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
|
||||
WSSKURI="^%sURI%sSSK%s%s$" % (OPTIONALHTTPLEAD, COLON, COLON, WSSKBITS)
|
||||
|
||||
WriteableSSKFileURI = StringConstraint(300, regexp=WSSKURI)
|
||||
|
||||
# NewDirectory Read-Write URI
|
||||
NDRWURI="^%sURI%sDIR2%s%s/?$" % (OPTIONALHTTPLEAD, COLON, COLON, WSSKBITS)
|
||||
|
||||
DirnodeURI = StringConstraint(300, regexp=NDRWURI)
|
||||
|
||||
MAX_BUCKETS = 200 # per peer
|
||||
|
||||
# MAX_SEGMENT_SIZE in encode.py is 1 MiB (this constraint allows k = 1)
|
||||
|
@ -168,7 +168,7 @@ class Invalid(unittest.TestCase):
|
||||
|
||||
class Constraint(unittest.TestCase):
|
||||
def test_constraint(self):
|
||||
good = 'URI:DIR:pb://xextf3eap44o3wi27mf7ehiur6wvhzr6@207.7.153.180:56677,127.0.0.1:56677/vdrive:qj51rfpnukhjmo7cm9awe5ks5e'
|
||||
good="http://127.0.0.1:8123/uri/URI%3ADIR2%3Aqo8ayna47cpw3rx3kho3mu7q4h%3Abk9qbgx76gh6eyj5ps8p6buz8fffw1ofc37e9w9d6ncsfpuz7icy/"
|
||||
DirnodeURI.checkObject(good, False)
|
||||
bad = good + '==='
|
||||
self.failUnlessRaises(Violation, DirnodeURI.checkObject, bad, False)
|
||||
@ -305,12 +305,12 @@ class NewDirnode(unittest.TestCase):
|
||||
|
||||
def test_is_string_newdirnode_rw(self):
|
||||
writekey = "\x01" * 16
|
||||
fingerprint = "\x02" * 16
|
||||
fingerprint = "\x02" * 32
|
||||
|
||||
n = uri.WriteableSSKFileURI(writekey, fingerprint)
|
||||
u1 = uri.NewDirectoryURI(n)
|
||||
|
||||
self.failUnless(uri.is_string_newdirnode_rw(u1.to_string()))
|
||||
self.failUnless(uri.is_string_newdirnode_rw(u1.to_string()), u1.to_string())
|
||||
|
||||
self.failIf(uri.is_string_newdirnode_rw("bogus"))
|
||||
self.failIf(uri.is_string_newdirnode_rw("URI:DIR2:bogus"))
|
||||
|
@ -5,7 +5,7 @@ from twisted.trial import unittest
|
||||
from twisted.internet import defer
|
||||
from twisted.web import client, error, http
|
||||
from twisted.python import failure, log
|
||||
from allmydata import webish, interfaces, provisioning
|
||||
from allmydata import interfaces, provisioning, uri, webish
|
||||
from allmydata.util import fileutil
|
||||
from allmydata.test.common import NonGridDirectoryNode, FakeCHKFileNode, FakeMutableFileNode, create_chk_filenode
|
||||
from allmydata.interfaces import IURI, INewDirectoryURI, IReadonlyNewDirectoryURI, IFileURI, IMutableFileURI, IMutableFileNode
|
||||
@ -1073,6 +1073,15 @@ class Web(WebMixin, unittest.TestCase):
|
||||
d.addCallback(self.failUnlessNodeKeysAre, [])
|
||||
return d
|
||||
|
||||
def test_POST_mkdir_no_parentdir(self):
|
||||
d = self.POST("/uri/", t="mkdir")
|
||||
def _after_mkdir(res):
|
||||
self.failUnless(uri.is_string_newdirnode_rw(res))
|
||||
# XXX TODO more
|
||||
d.addCallback(_after_mkdir)
|
||||
return d
|
||||
test_POST_mkdir_no_parentdir.skip = "not yet implemented"
|
||||
|
||||
def test_POST_mkdir_replace(self): # return value?
|
||||
d = self.POST(self.public_url + "/foo", t="mkdir", name="sub")
|
||||
d.addCallback(lambda res: self._foo_node.get("sub"))
|
||||
|
@ -4,7 +4,8 @@ from zope.interface import implements
|
||||
from twisted.python.components import registerAdapter
|
||||
from allmydata.util import idlib, hashutil
|
||||
from allmydata.interfaces import IURI, IDirnodeURI, IFileURI, IVerifierURI, \
|
||||
IMutableFileURI, INewDirectoryURI, IReadonlyNewDirectoryURI
|
||||
IMutableFileURI, INewDirectoryURI, IReadonlyNewDirectoryURI, DirnodeURI
|
||||
import foolscap
|
||||
|
||||
# the URI shall be an ascii representation of the file. It shall contain
|
||||
# enough information to retrieve and validate the contents. It shall be
|
||||
@ -417,13 +418,11 @@ def from_string_dirnode(s):
|
||||
registerAdapter(from_string_dirnode, str, IDirnodeURI)
|
||||
|
||||
def is_string_newdirnode_rw(s):
|
||||
if not s.startswith("URI:DIR2:"):
|
||||
return False
|
||||
try:
|
||||
(header_uri, header_dir2, writekey_s, fingerprint_s) = s.split(":", 3)
|
||||
except ValueError:
|
||||
DirnodeURI.checkObject(s, inbound=False)
|
||||
return True
|
||||
except foolscap.tokens.Violation, v:
|
||||
return False
|
||||
return idlib.could_be_base32_encoded(writekey_s) and idlib.could_be_base32_encoded(fingerprint_s)
|
||||
|
||||
def from_string_filenode(s):
|
||||
u = from_string(s)
|
||||
|
Loading…
x
Reference in New Issue
Block a user