strawmelonjuice.com
Forking what is left of ZeroNet and hopefully adding an AT Proto Frontend/Proxy
1import io
2
3import pytest
4import time
5
6from Connection import ConnectionServer
7from Connection import Connection
8from File import FileServer
9
10
11@pytest.mark.usefixtures("resetSettings")
12@pytest.mark.usefixtures("resetTempSettings")
13class TestFileRequest:
14 def testGetFile(self, file_server, site):
15 file_server.ip_incoming = {} # Reset flood protection
16 client = ConnectionServer(file_server.ip, 1545)
17
18 connection = client.getConnection(file_server.ip, 1544)
19 file_server.sites[site.address] = site
20
21 # Normal request
22 response = connection.request("getFile", {"site": site.address, "inner_path": "content.json", "location": 0})
23 assert b"sign" in response["body"]
24
25 response = connection.request("getFile", {"site": site.address, "inner_path": "content.json", "location": 0, "file_size": site.storage.getSize("content.json")})
26 assert b"sign" in response["body"]
27
28 # Invalid file
29 response = connection.request("getFile", {"site": site.address, "inner_path": "invalid.file", "location": 0})
30 assert "File read error" in response["error"]
31
32 # Location over size
33 response = connection.request("getFile", {"site": site.address, "inner_path": "content.json", "location": 1024 * 1024})
34 assert "File read error" in response["error"]
35
36 # Stream from parent dir
37 response = connection.request("getFile", {"site": site.address, "inner_path": "../users.json", "location": 0})
38 assert "File read exception" in response["error"]
39
40 # Invalid site
41 response = connection.request("getFile", {"site": "", "inner_path": "users.json", "location": 0})
42 assert "Unknown site" in response["error"]
43
44 response = connection.request("getFile", {"site": ".", "inner_path": "users.json", "location": 0})
45 assert "Unknown site" in response["error"]
46
47 # Invalid size
48 response = connection.request("getFile", {"site": site.address, "inner_path": "content.json", "location": 0, "file_size": 1234})
49 assert "File size does not match" in response["error"]
50
51 # Invalid path
52 for path in ["../users.json", "./../users.json", "data/../content.json", ".../users.json"]:
53 for sep in ["/", "\\"]:
54 response = connection.request("getFile", {"site": site.address, "inner_path": path.replace("/", sep), "location": 0})
55 assert response["error"] == 'File read exception'
56
57 connection.close()
58 client.stop()
59
60 def testStreamFile(self, file_server, site):
61 file_server.ip_incoming = {} # Reset flood protection
62 client = ConnectionServer(file_server.ip, 1545)
63 connection = client.getConnection(file_server.ip, 1544)
64 file_server.sites[site.address] = site
65
66 buff = io.BytesIO()
67 response = connection.request("streamFile", {"site": site.address, "inner_path": "content.json", "location": 0}, buff)
68 assert "stream_bytes" in response
69 assert b"sign" in buff.getvalue()
70
71 # Invalid file
72 buff = io.BytesIO()
73 response = connection.request("streamFile", {"site": site.address, "inner_path": "invalid.file", "location": 0}, buff)
74 assert "File read error" in response["error"]
75
76 # Location over size
77 buff = io.BytesIO()
78 response = connection.request(
79 "streamFile", {"site": site.address, "inner_path": "content.json", "location": 1024 * 1024}, buff
80 )
81 assert "File read error" in response["error"]
82
83 # Stream from parent dir
84 buff = io.BytesIO()
85 response = connection.request("streamFile", {"site": site.address, "inner_path": "../users.json", "location": 0}, buff)
86 assert "File read exception" in response["error"]
87
88 connection.close()
89 client.stop()
90
91 def testPex(self, file_server, site, site_temp):
92 file_server.sites[site.address] = site
93 client = FileServer(file_server.ip, 1545)
94 client.sites = {site_temp.address: site_temp}
95 site_temp.connection_server = client
96 connection = client.getConnection(file_server.ip, 1544)
97
98 # Add new fake peer to site
99 fake_peer = site.addPeer(file_server.ip_external, 11337, return_peer=True)
100 # Add fake connection to it
101 fake_peer.connection = Connection(file_server, file_server.ip_external, 11337)
102 fake_peer.connection.last_recv_time = time.time()
103 assert fake_peer in site.getConnectablePeers()
104
105 # Add file_server as peer to client
106 peer_file_server = site_temp.addPeer(file_server.ip, 1544)
107
108 assert "%s:11337" % file_server.ip_external not in site_temp.peers
109 assert peer_file_server.pex()
110 assert "%s:11337" % file_server.ip_external in site_temp.peers
111
112 # Should not exchange private peers from local network
113 fake_peer_private = site.addPeer("192.168.0.1", 11337, return_peer=True)
114 assert fake_peer_private not in site.getConnectablePeers(allow_private=False)
115 fake_peer_private.connection = Connection(file_server, "192.168.0.1", 11337)
116 fake_peer_private.connection.last_recv_time = time.time()
117
118 assert "192.168.0.1:11337" not in site_temp.peers
119 assert not peer_file_server.pex()
120 assert "192.168.0.1:11337" not in site_temp.peers
121
122
123 connection.close()
124 client.stop()