/
servertests.py
151 lines (128 loc) · 5.27 KB
/
servertests.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
#!/usr/bin/python3 -OO
# Copyright 2007-2022 The SABnzbd-Team <team@sabnzbd.org>
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
"""
sabnzbd.utils.servertests - Debugging server connections. Currently only NNTP server tests are done.
"""
import socket
import sys
from sabnzbd.constants import DEF_TIMEOUT
from sabnzbd.newswrapper import NewsWrapper, NNTPPermanentError
from sabnzbd.downloader import Server, clues_login, clues_too_many, nntp_to_msg
from sabnzbd.config import get_servers
from sabnzbd.misc import int_conv, match_str
def test_nntp_server_dict(kwargs):
"""Will connect (blocking) to the NNTP server and report back any errors"""
host = kwargs.get("host", "").strip()
port = int_conv(kwargs.get("port", 0))
username = kwargs.get("username", "").strip()
password = kwargs.get("password", "").strip()
server = kwargs.get("server", "").strip()
connections = int_conv(kwargs.get("connections", 0))
timeout = int_conv(kwargs.get("timeout", DEF_TIMEOUT))
ssl = int_conv(kwargs.get("ssl", 0))
ssl_verify = int_conv(kwargs.get("ssl_verify", 1))
ssl_ciphers = kwargs.get("ssl_ciphers", "").strip()
if not host:
return False, T("The hostname is not set.")
if not connections:
return False, T("There are no connections set. Please set at least one connection.")
if not port:
if ssl:
port = 563
else:
port = 119
if not timeout:
# Lower value during new server testing
timeout = 10
if "*" in password and not password.strip("*"):
# If the password is masked, try retrieving it from the config
srv = get_servers().get(server)
if srv:
password = srv.password()
else:
return False, T("Password masked in ******, please re-enter")
try:
s = Server(
server_id=-1,
displayname="",
host=host,
port=port,
timeout=timeout,
threads=0,
priority=0,
use_ssl=ssl,
ssl_verify=ssl_verify,
ssl_ciphers=ssl_ciphers,
send_group=False,
username=username,
password=password,
)
except:
return False, T("Invalid server details")
try:
nw = NewsWrapper(server=s, thrdnum=-1, block=True)
nw.init_connect()
while not nw.connected:
nw.clear_data()
nw.recv_chunk(block=True)
nw.finish_connect(nw.status_code)
except socket.timeout:
if port != 119 and not ssl:
return False, T("Timed out: Try enabling SSL or connecting on a different port.")
else:
return False, T("Timed out")
except socket.error as err:
# Trying SSL on non-SSL port?
if match_str(str(err), ("unknown protocol", "wrong version number")):
return False, T("Unknown SSL protocol: Try disabling SSL or connecting on a different port.")
return False, str(err)
except TypeError:
return False, T("Invalid server address.")
except IndexError:
# No data was received in recv_chunk() call
return False, T("Server quit during login sequence.")
except NNTPPermanentError:
# Handled by the code below
pass
except Exception as err:
return False, str(err)
if not username or not password:
nw.nntp.sock.sendall(b"ARTICLE <test@home>\r\n")
try:
nw.clear_data()
nw.recv_chunk(block=True)
except:
# Some internal error, not always safe to close connection
return False, str(sys.exc_info()[1])
# Parse result
return_status = ()
if nw.status_code:
if nw.status_code == 480:
return_status = (False, T("Server requires username and password."))
elif nw.status_code < 300 or nw.status_code in (411, 423, 430):
# If no username/password set and we requested fake-article, it will return 430 Not Found
return_status = (True, T("Connection Successful!"))
elif nw.status_code == 502 or clues_login(nntp_to_msg(nw.data)):
return_status = (False, T("Authentication failed, check username/password."))
elif clues_too_many(nntp_to_msg(nw.data)):
return_status = (False, T("Too many connections, please pause downloading or try again later"))
# Fallback in case no data was received or unknown status
if not return_status:
return_status = (False, T("Could not determine connection result (%s)") % nntp_to_msg(nw.data))
# Close the connection and return result
nw.hard_reset(send_quit=True)
return return_status