/
daemon_get_nodeinfo.py
151 lines (143 loc) · 5.64 KB
/
daemon_get_nodeinfo.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
__filename__ = "daemon_get_nodeinfo.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "1.5.0"
__maintainer__ = "Bob Mottram"
__email__ = "bob@libreserver.org"
__status__ = "Production"
__module_group__ = "Core GET"
import json
from httpcodes import http_400
from httpcodes import http_404
from httpcodes import http_503
from httpcodes import write2
from httpheaders import set_headers
from utils import convert_domains
from utils import get_instance_url
from utils import local_network_host
from siteactive import referer_is_active
from crawlers import update_known_crawlers
from blocking import broch_mode_is_active
from metadata import meta_data_node_info
def get_nodeinfo(self, ua_str: str, calling_domain: str,
referer_domain: str,
http_prefix: str, calling_site_timeout: int,
debug: bool, base_dir: str,
unit_test: bool,
domain_full: str,
path: str,
allow_local_network_access: bool,
sites_unavailable: [],
known_crawlers: [],
onion_domain: str,
i2p_domain: str,
project_version: str,
show_node_info_version: bool,
show_node_info_accounts: bool,
registration: bool, domain: str) -> bool:
if path.startswith('/nodeinfo/1.0'):
http_400(self)
return True
if not path.startswith('/nodeinfo/2.'):
return False
if not referer_domain:
if not debug and not unit_test:
print('nodeinfo request has no referer domain ' + str(ua_str))
http_400(self)
return True
if referer_domain == domain_full:
print('nodeinfo request from self')
http_400(self)
return True
if self.server.nodeinfo_is_active:
if not referer_domain:
print('nodeinfo is busy during request without referer domain')
else:
print('nodeinfo is busy during request from ' + referer_domain)
http_503(self)
return True
self.server.nodeinfo_is_active = True
# is this a real website making the call ?
if not debug and not unit_test and referer_domain:
# Does calling_domain look like a domain?
if ' ' in referer_domain or \
';' in referer_domain or \
'.' not in referer_domain:
print('nodeinfo referer domain does not look like a domain ' +
referer_domain)
http_400(self)
self.server.nodeinfo_is_active = False
return True
if not allow_local_network_access:
if local_network_host(referer_domain):
print('nodeinfo referer domain is from the ' +
'local network ' + referer_domain)
http_400(self)
self.server.nodeinfo_is_active = False
return True
if not referer_is_active(http_prefix,
referer_domain, ua_str,
calling_site_timeout,
sites_unavailable):
print('nodeinfo referer url is not active ' +
referer_domain)
http_400(self)
self.server.nodeinfo_is_active = False
return True
if debug:
print('DEBUG: nodeinfo ' + path)
crawl_time = \
update_known_crawlers(ua_str,
base_dir,
known_crawlers,
self.server.last_known_crawler)
if crawl_time is not None:
self.server.last_known_crawler = crawl_time
# If we are in broch mode then don't show potentially
# sensitive metadata.
# For example, if this or allied instances are being attacked
# then numbers of accounts may be changing as people
# migrate, and that information may be useful to an adversary
broch_mode = broch_mode_is_active(base_dir)
node_info_version = project_version
if not show_node_info_version or broch_mode:
node_info_version = '0.0.0'
if broch_mode:
show_node_info_accounts = False
instance_url = get_instance_url(calling_domain,
http_prefix,
domain_full,
onion_domain,
i2p_domain)
about_url = instance_url + '/about'
terms_of_service_url = instance_url + '/terms'
info = meta_data_node_info(base_dir,
about_url, terms_of_service_url,
registration,
node_info_version,
show_node_info_accounts,
domain)
if info:
msg_str = json.dumps(info)
msg_str = convert_domains(calling_domain, referer_domain,
msg_str, http_prefix,
domain,
onion_domain,
i2p_domain)
msg = msg_str.encode('utf-8')
msglen = len(msg)
protocol_str = \
'application/json; profile=' + \
'"http://nodeinfo.diaspora.software/ns/schema/2.1#"'
set_headers(self, protocol_str, msglen,
None, calling_domain, True)
write2(self, msg)
if referer_domain:
print('nodeinfo sent to ' + referer_domain)
else:
print('nodeinfo sent to unknown referer')
self.server.nodeinfo_is_active = False
return True
http_404(self, 5)
self.server.nodeinfo_is_active = False
return True