Fixed path-disclosure on connectionParameters

src/server.py

@@ -50,6 +50,13 @@ class index(object):
     def GET(self, format):
         return "This is MRA."
 
+class tests(object):
+    def PUT(self, name, format):
+        spath = tools.safe_path_join(get_config('storage')['mapfiles'], "%s.map" % get_config("testing")["model"])
+        tpath = tools.safe_path_join(get_config('storage')['mapfiles'], "%s.map" % name)
+        open(tpath, "w").write(open(spath).read())
+
+        webapp.Created("%s/maps/%s.%s" % (web.ctx.home, name, format))
 
 class mapfiles(object):
     @HTTPCompatible()
@@ -479,7 +486,7 @@ class files(object):
                 z.extract(f, path=tools.get_st_data_path(ws_name, st_type, st_name))
 
         # Set new connection parameters:
-        ws.update_store(st_type, st_name, {"connectionParameters":{"path":path}})
+        ws.update_store(st_type, st_name, {"connectionParameters":{"url":"file:"+tools.no_res_root(path)}})
         ws.save()
 
         # Finally we might have to configure it.
@@ -851,6 +858,10 @@ class layergroup(object):
 # Index:
 urlmap(index, "")
 
+# Tests
+if get_config("testing")["active"]:
+    urlmap(tests, "tests", ())
+
 # Styler: TODO
 #urlmap(styler, format = False)
 

src/tools.py

@@ -131,6 +131,14 @@ def mk_style_path(name, *args):
     mk_path(path)
     return path
 
+def no_root(root, path):
+    path = os.path.abspath(path)
+    root = os.path.abspath(root)
+    return path[len(root):] if path.startswith(root) else path
+
+def no_res_root(path):
+    return no_root(get_config('storage')['resources'], path)
+
 def is_hidden(path):
     # TODO Add a lot of checks, recursive option (to check folders)
     # MacOSX has at least four ways to hide files...

tests/testScenario.py

@@ -28,15 +28,18 @@ from utils import APIRequest
 
 import sys
 
-def testScenario():
 
-    target = "http://192.168.1.39/mra/"
-    map_name = "test"
+def test_scenario():
 
+    target = "http://localhost:8080"
+    map_name = "tests"
+
+    # Clean the test file, now we are sure it is empty.
+    APIRequest("PUT", target + "/tests/" + map_name)
 
     # GET workspaces.
 
-    wss = APIRequest("GET", target + "/maps/test/workspaces")["workspaces"]
+    wss = APIRequest("GET", target + "/maps/" + map_name + "/workspaces")["workspaces"]
     assert len(wss) == 1
     assert wss[0]["name"] == "default"
 
@@ -48,11 +51,7 @@ def testScenario():
     # GET dataStores
 
     dss = APIRequest("GET", ws["dataStores"]["href"])["dataStores"]
-
-    # DELETE the dataStores
-
-    for ds in dss:
-        APIRequest("DELETE", ds["href"])
+    assert len(dss) == 0
 
     # POST a datastore and GET it
 
@@ -80,12 +79,22 @@ def testScenario():
     assert len(fts) == 0
 
 
-    # PUT file
+    # PUT file, and check if datastore is updated.
 
     APIRequest("PUT", ds_link + "/file.shp", open("./files/timezones_shp.zip", "rb"),
                encode=None, content_type="application/zip")
 
     ds = APIRequest("GET", ds_link)["dataStore"]
-    assert ds["connectionParameters"]["path"] = "workspaces/%s/datastores/%s/timezones.shp" % (
-        ws["name"], ds["name"])
+    assert ds["connectionParameters"]["url"] == "file:/workspaces/%s/datastores/%s/timezones.shp" % (ws["name"], ds["name"])
+
+    # POST a featuretype
+
+    name, title = "testFT1", "test feature type 1"
+    _, r = APIRequest("POST", ds["href"], {"featureType":{"name":name, "title":title}},
+                      get_response=True)
+    ft_link = r.getheader("Location")
+
+    ft = APIRequest("GET", ft_link)["featureType"]
+    assert ft["name"] == name
+    assert ds["title"] == title
 

tests/utils.py

@@ -52,8 +52,10 @@ def APIRequest(method, url, data=None, encode="json", decode="json", content_typ
 
     if encode and not url.endswith("." + encode):
         url = surl.path + "." + encode
+    else:
+        url = surl.path
 
-    print >>sys.stderr, method, url
+    print >>sys.stderr, method, surl.geturl().replace(surl.path, url)
     conn = httplib.HTTPConnection(surl.hostname, surl.port)
     conn.request(method, url, body=data, headers={"Content-Type":content_type})
 
@@ -62,6 +64,7 @@ def APIRequest(method, url, data=None, encode="json", decode="json", content_typ
     if expected_type == None:
         expected_type = deduce_content_type(decode)
 
+    # TODO: enable this test once it is suported.
     # assert expected_type in r.getheader("Content-Type"), "received %s instead of %s" % (
     #     r.getheader("Content-Type"), expected_type)