暂无描述
您最多选择25个主题 主题必须以字母或数字开头,可以包含连字符 (-),并且长度不得超过35个字符

tests.py 13KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392
  1. import unittest
  2. from time import sleep
  3. import uuid
  4. import socket
  5. import requests
  6. import os
  7. TEST_SERVER = 'sovereign.local'
  8. TEST_ADDRESS = 'sovereign@sovereign.local'
  9. TEST_PASSWORD = 'foo'
  10. CA_BUNDLE = 'roles/common/files/wildcard_ca.pem'
  11. socket.setdefaulttimeout(5)
  12. os.environ['REQUESTS_CA_BUNDLE'] = CA_BUNDLE
  13. class SSHTests(unittest.TestCase):
  14. def test_ssh_banner(self):
  15. """SSH is responding with its banner"""
  16. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  17. s.connect((TEST_SERVER, 22))
  18. data = s.recv(1024)
  19. s.close()
  20. self.assertRegexpMatches(data, '^SSH-2.0-OpenSSH')
  21. class WebTests(unittest.TestCase):
  22. def test_blog_http(self):
  23. """Blog is redirecting to https"""
  24. # FIXME: requests won't verify sovereign.local with *.sovereign.local cert
  25. r = requests.get('http://' + TEST_SERVER, verify=False)
  26. # We should be redirected to https
  27. self.assertEquals(r.history[0].status_code, 301)
  28. self.assertEquals(r.url, 'https://' + TEST_SERVER + '/')
  29. # 403 - Since there is no documents in the blog directory
  30. self.assertEquals(r.status_code, 403)
  31. def test_webmail_http(self):
  32. """Webmail is redirecting to https and displaying login page"""
  33. r = requests.get('http://mail.' + TEST_SERVER)
  34. # We should be redirected to https
  35. self.assertEquals(r.history[0].status_code, 301)
  36. self.assertEquals(r.url, 'https://mail.' + TEST_SERVER + '/')
  37. # 200 - We should be at the login page
  38. self.assertEquals(r.status_code, 200)
  39. self.assertIn(
  40. 'Welcome to Roundcube Webmail',
  41. r.content
  42. )
  43. def test_zpush_http_unauthorized(self):
  44. r = requests.get('http://mail.' + TEST_SERVER + '/Microsoft-Server-ActiveSync')
  45. # We should be redirected to https
  46. self.assertEquals(r.history[0].status_code, 301)
  47. self.assertEquals(r.url, 'https://mail.' + TEST_SERVER + '/Microsoft-Server-ActiveSync')
  48. # Unauthorized
  49. self.assertEquals(r.status_code, 401)
  50. def test_zpush_https(self):
  51. r = requests.post('https://mail.' + TEST_SERVER + '/Microsoft-Server-ActiveSync',
  52. auth=('sovereign@sovereign.local', 'foo'),
  53. params={
  54. 'DeviceId': '1234',
  55. 'DeviceType': 'testbot',
  56. 'Cmd': 'Ping',
  57. })
  58. self.assertEquals(r.headers['content-type'],
  59. 'application/vnd.ms-sync.wbxml')
  60. self.assertEquals(r.headers['ms-server-activesync'],
  61. '14.0')
  62. def test_owncloud_http(self):
  63. """ownCloud is redirecting to https and displaying login page"""
  64. r = requests.get('http://cloud.' + TEST_SERVER)
  65. # We should be redirected to https
  66. self.assertEquals(r.history[0].status_code, 301)
  67. self.assertEquals(r.url, 'https://cloud.' + TEST_SERVER + '/')
  68. # 200 - We should be at the login page
  69. self.assertEquals(r.status_code, 200)
  70. self.assertIn(
  71. 'ownCloud',
  72. r.content
  73. )
  74. def test_selfoss_http(self):
  75. """selfoss is redirecting to https and displaying login page"""
  76. r = requests.get('http://news.' + TEST_SERVER)
  77. # We should be redirected to https
  78. self.assertEquals(r.history[0].status_code, 301)
  79. self.assertEquals(r.url, 'https://news.' + TEST_SERVER + '/')
  80. # 200 - We should be at the login page
  81. self.assertEquals(r.status_code, 200)
  82. self.assertIn(
  83. 'selfoss',
  84. r.content
  85. )
  86. self.assertIn(
  87. 'login',
  88. r.content
  89. )
  90. def test_cgit_http(self):
  91. """CGit web interface is displaying home page"""
  92. r = requests.get('http://git.' + TEST_SERVER, verify=False)
  93. # We should be redirected to https
  94. self.assertEquals(r.history[0].status_code, 301)
  95. self.assertEquals(r.url, 'https://git.' + TEST_SERVER + '/')
  96. # 200 - We should be at the repository page
  97. self.assertEquals(r.status_code, 200)
  98. self.assertIn(
  99. 'git repository',
  100. r.content
  101. )
  102. class IRCTests(unittest.TestCase):
  103. def test_irc_auth(self):
  104. """ZNC is accepting encrypted logins"""
  105. import ssl
  106. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  107. ssl_sock = ssl.wrap_socket(s, ca_certs=CA_BUNDLE, cert_reqs=ssl.CERT_REQUIRED)
  108. ssl_sock.connect((TEST_SERVER, 6697))
  109. # Check the encryption parameters
  110. cipher, version, bits = ssl_sock.cipher()
  111. self.assertEquals(cipher, 'AES256-GCM-SHA384')
  112. self.assertEquals(version, 'TLSv1/SSLv3')
  113. self.assertEquals(bits, 256)
  114. # Login
  115. ssl_sock.send('CAP REQ sasl multi-prefix\r\n')
  116. ssl_sock.send('PASS foo\r\n')
  117. ssl_sock.send('NICK sovereign\r\n')
  118. ssl_sock.send('USER sovereign 0 * Sov\r\n')
  119. # Read until we see the ZNC banner (or timeout)
  120. while 1:
  121. r = ssl_sock.recv(1024)
  122. if 'Connected to ZNC' in r:
  123. break
  124. def new_message(from_email, to_email):
  125. """Creates an email (headers & body) with a random subject"""
  126. from email.mime.text import MIMEText
  127. msg = MIMEText('Testing')
  128. msg['Subject'] = uuid.uuid4().hex[:8]
  129. msg['From'] = from_email
  130. msg['To'] = to_email
  131. return msg.as_string(), msg['subject']
  132. class MailTests(unittest.TestCase):
  133. def assertIMAPReceived(self, subject):
  134. """Connects with IMAP and asserts the existence of an email, then deletes it"""
  135. import imaplib
  136. sleep(1)
  137. # Login to IMAP
  138. m = imaplib.IMAP4_SSL(TEST_SERVER, 993)
  139. m.login(TEST_ADDRESS, TEST_PASSWORD)
  140. m.select()
  141. # Assert the message exists
  142. typ, data = m.search(None, '(SUBJECT \"{}\")'.format(subject))
  143. self.assertTrue(len(data[0].split()), 1)
  144. # Delete it & logout
  145. m.store(data[0].strip(), '+FLAGS', '\\Deleted')
  146. m.expunge()
  147. m.close()
  148. m.logout()
  149. def assertPOP3Received(self, subject):
  150. """Connects with POP3S and asserts the existence of an email, then deletes it"""
  151. import poplib
  152. sleep(1)
  153. # Login to POP3
  154. mail = poplib.POP3_SSL(TEST_SERVER, 995)
  155. mail.user(TEST_ADDRESS)
  156. mail.pass_(TEST_PASSWORD)
  157. # Assert the message exists
  158. num = len(mail.list()[1])
  159. resp, text, octets = mail.retr(num)
  160. self.assertTrue("Subject: " + subject in text)
  161. # Delete it and log out
  162. mail.dele(num)
  163. mail.quit()
  164. def test_imap_requires_ssl(self):
  165. """IMAP without SSL is NOT available"""
  166. import imaplib
  167. with self.assertRaisesRegexp(socket.timeout, 'timed out'):
  168. imaplib.IMAP4(TEST_SERVER, 143)
  169. def test_pop3_requires_ssl(self):
  170. """POP3 without SSL is NOT available"""
  171. import poplib
  172. with self.assertRaisesRegexp(socket.timeout, 'timed out'):
  173. poplib.POP3(TEST_SERVER, 110)
  174. def test_smtps(self):
  175. """Email sent from an MUA via SMTPS is delivered"""
  176. import smtplib
  177. msg, subject = new_message(TEST_ADDRESS, 'root@sovereign.local')
  178. s = smtplib.SMTP_SSL(TEST_SERVER, 465)
  179. s.login(TEST_ADDRESS, TEST_PASSWORD)
  180. s.sendmail(TEST_ADDRESS, ['root@sovereign.local'], msg)
  181. s.quit()
  182. self.assertIMAPReceived(subject)
  183. def test_smtps_delimiter_to(self):
  184. """Email sent to address with delimiter is delivered"""
  185. import smtplib
  186. msg, subject = new_message(TEST_ADDRESS, 'root+foo@sovereign.local')
  187. s = smtplib.SMTP_SSL(TEST_SERVER, 465)
  188. s.login(TEST_ADDRESS, TEST_PASSWORD)
  189. s.sendmail(TEST_ADDRESS, ['root+foo@sovereign.local'], msg)
  190. s.quit()
  191. self.assertIMAPReceived(subject)
  192. def test_smtps_requires_auth(self):
  193. """SMTPS with no authentication is rejected"""
  194. import smtplib
  195. s = smtplib.SMTP_SSL(TEST_SERVER, 465)
  196. with self.assertRaisesRegexp(smtplib.SMTPRecipientsRefused, 'Access denied'):
  197. s.sendmail(TEST_ADDRESS, ['root@sovereign.local'], 'Test')
  198. s.quit()
  199. def test_smtp(self):
  200. """Email sent from an MTA is delivered"""
  201. import smtplib
  202. msg, subject = new_message('someone@example.com', TEST_ADDRESS)
  203. s = smtplib.SMTP(TEST_SERVER, 25)
  204. s.sendmail('someone@example.com', [TEST_ADDRESS], msg)
  205. s.quit()
  206. self.assertIMAPReceived(subject)
  207. def test_smtp_tls(self):
  208. """Email sent from an MTA via SMTP+TLS is delivered"""
  209. import smtplib
  210. msg, subject = new_message('someone@example.com', TEST_ADDRESS)
  211. s = smtplib.SMTP(TEST_SERVER, 25)
  212. s.starttls()
  213. s.sendmail('someone@example.com', [TEST_ADDRESS], msg)
  214. s.quit()
  215. self.assertIMAPReceived(subject)
  216. def test_smtps_headers(self):
  217. """Email sent from an MUA has DKIM and TLS headers"""
  218. import smtplib
  219. import imaplib
  220. # Send a message to root
  221. msg, subject = new_message(TEST_ADDRESS, 'root@sovereign.local')
  222. s = smtplib.SMTP_SSL(TEST_SERVER, 465)
  223. s.login(TEST_ADDRESS, TEST_PASSWORD)
  224. s.sendmail(TEST_ADDRESS, ['root@sovereign.local'], msg)
  225. s.quit()
  226. sleep(1)
  227. # Get the message
  228. m = imaplib.IMAP4_SSL(TEST_SERVER, 993)
  229. m.login(TEST_ADDRESS, TEST_PASSWORD)
  230. m.select()
  231. _, res = m.search(None, '(SUBJECT \"{}\")'.format(subject))
  232. _, data = m.fetch(res[0], '(RFC822)')
  233. self.assertIn(
  234. 'DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=sovereign.local;',
  235. data[0][1]
  236. )
  237. self.assertIn(
  238. 'ECDHE-RSA-AES256-GCM-SHA384 (256/256 bits)',
  239. data[0][1]
  240. )
  241. # Clean up
  242. m.store(res[0].strip(), '+FLAGS', '\\Deleted')
  243. m.expunge()
  244. m.close()
  245. m.logout()
  246. def test_smtp_headers(self):
  247. """Email sent from an MTA via SMTP+TLS has X-DSPAM and TLS headers"""
  248. import smtplib
  249. import imaplib
  250. # Send a message to root
  251. msg, subject = new_message('someone@example.com', TEST_ADDRESS)
  252. s = smtplib.SMTP(TEST_SERVER, 25)
  253. s.starttls()
  254. s.sendmail('someone@example.com', [TEST_ADDRESS], msg)
  255. s.quit()
  256. sleep(1)
  257. # Get the message
  258. m = imaplib.IMAP4_SSL(TEST_SERVER, 993)
  259. m.login(TEST_ADDRESS, TEST_PASSWORD)
  260. m.select()
  261. _, res = m.search(None, '(SUBJECT \"{}\")'.format(subject))
  262. _, data = m.fetch(res[0], '(RFC822)')
  263. self.assertIn(
  264. 'X-DSPAM-Result: ',
  265. data[0][1]
  266. )
  267. self.assertIn(
  268. 'ECDHE-RSA-AES256-GCM-SHA384 (256/256 bits)',
  269. data[0][1]
  270. )
  271. # Clean up
  272. m.store(res[0].strip(), '+FLAGS', '\\Deleted')
  273. m.expunge()
  274. m.close()
  275. m.logout()
  276. def test_pop3s(self):
  277. """Connects with POP3S and asserts the existance of an email, then deletes it"""
  278. import smtplib
  279. msg, subject = new_message(TEST_ADDRESS, 'root@sovereign.local')
  280. s = smtplib.SMTP_SSL(TEST_SERVER, 465)
  281. s.login(TEST_ADDRESS, TEST_PASSWORD)
  282. s.sendmail(TEST_ADDRESS, ['root@sovereign.local'], msg)
  283. s.quit()
  284. self.assertPOP3Received(subject)
  285. class XMPPTests(unittest.TestCase):
  286. def test_xmpp_c2s(self):
  287. """Prosody is listening on 5222 for clients and requiring TLS"""
  288. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  289. s.connect((TEST_SERVER, 5222))
  290. # Based off http://wiki.xmpp.org/web/Programming_Jabber_Clients
  291. s.send("<stream:stream xmlns:stream='http://etherx.jabber.org/streams' "
  292. "xmlns='jabber:client' to='sovereign.local' version='1.0'>")
  293. data = s.recv(1024)
  294. s.close()
  295. self.assertRegexpMatches(
  296. data,
  297. "<starttls xmlns='urn:ietf:params:xml:ns:xmpp-tls'><required/></starttls>"
  298. )
  299. def test_xmpp_s2s(self):
  300. """Prosody is listening on 5269 for servers"""
  301. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  302. s.connect((TEST_SERVER, 5269))
  303. # Base off http://xmpp.org/extensions/xep-0114.html
  304. s.send("<stream:stream xmlns:stream='http://etherx.jabber.org/streams' "
  305. "xmlns='jabber:component:accept' to='sovereign.local'>")
  306. data = s.recv(1024)
  307. s.close()
  308. self.assertRegexpMatches(
  309. data,
  310. "from='sovereign.local'"
  311. )