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
|
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# 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.
"""The suite must pass with no network, including the network modules.
A network module that can only be tested with a network is a module that
stops being tested. rdap and contacts both take an injected fetch, and
this asserts that the default is never reached by accident during a test
run: parse stays pure, and nothing above it opens a socket unasked.
"""
import socket
import unittest
from unittest import mock
from abusectl import contacts, parse, rdap
class NothingOpensASocket(unittest.TestCase):
def setUp(self):
# socket.socket.connect, NOT socket.socket: replacing the class
# itself breaks the ssl module at import time and produces false
# failures that have nothing to do with network use.
patcher = mock.patch.object(
socket.socket, "connect",
side_effect=AssertionError("socket.socket.connect was called"),
)
patcher.start()
self.addCleanup(patcher.stop)
for name in ("create_connection", "getaddrinfo"):
patcher = mock.patch.object(
socket, name,
side_effect=AssertionError(f"socket.{name} was called"),
)
patcher.start()
self.addCleanup(patcher.stop)
def test_parsing_opens_no_socket(self):
raw = (b"Received: from relay.example.invalid ([192.0.2.10])\r\n"
b"From: sender@example.invalid\r\n"
b"Subject: test\r\n\r\nbody\r\n")
parse.iocs(raw, trusted=["192.0.2.0/24"])
def test_selecting_report_headers_opens_no_socket(self):
# A new entry point into the parse path, so it is held to the same
# guarantee: choosing what to publish resolves nothing.
raw = (b"Received: from relay.example.invalid ([192.0.2.10])\r\n"
b"From: sender@example.invalid\r\n"
b"Subject: test\r\n\r\nbody\r\n")
parse.report_headers(raw, trusted=["192.0.2.0/24"])
def test_resolving_with_an_injected_fetch_opens_no_socket(self):
iocs = [{"id": "ioc-1", "type": "ipv4", "value": "198.51.100.7"}]
bootstraps = {
"ipv4": {"services": [[["198.51.100.0/24"],
["https://rir.example.invalid/"]]]},
"ipv6": {"services": []},
"dns": {"services": []},
}
result = contacts.resolve(
iocs, bootstraps=bootstraps,
fetch=lambda url: {"handle": "NET-1", "entities": []},
)
# Assert on the handle the injected fetch produced, not merely on
# the length. resolve() records a transport failure as a per-entry
# error rather than raising, so an entry exists either way: a test
# counting entries would stay green if the injection were removed
# and the real transport were blocked, which is the one thing this
# test is here to notice.
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["handle"], "NET-1")
def test_the_real_transport_is_never_the_default_in_a_test(self):
"""Sanity: http_fetch exists and is the documented default."""
self.assertIs(contacts.resolve.__defaults__[-1], rdap.http_fetch)
if __name__ == "__main__":
unittest.main()
|
