diff --git a/tests/conftest.py b/tests/conftest.py index c4167b9..4adf233 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,4 +1,5 @@ import pytest +import base64 from jail2ban import create_app @@ -26,3 +27,8 @@ def client(app): @pytest.fixture() def runner(app): return app.test_cli_runner() + + +@pytest.fixture() +def valid_credentials(): + return base64.b64encode(b"test.example.com:testpassword").decode("utf-8") diff --git a/tests/test_ban.py b/tests/test_ban.py index 4da17de..a2553f2 100644 --- a/tests/test_ban.py +++ b/tests/test_ban.py @@ -1,8 +1,7 @@ -import base64 from types import SimpleNamespace -def test_ban_ipv6(client, mocker): +def test_ban_ipv6(client, mocker, valid_credentials): def noop(): pass run_res = SimpleNamespace() @@ -13,16 +12,16 @@ def test_ban_ipv6(client, mocker): mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"name": "sshd", "ip": "2001:db8::abad:cafe"} response = client.put("/ban", json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert response.json['operation'] == 'add' -def test_ban_ipv4(client, mocker): +def test_ban_ipv4(client, mocker, valid_credentials): def noop(): pass run_res = SimpleNamespace() @@ -33,15 +32,16 @@ def test_ban_ipv4(client, mocker): mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"name": "sshd", "ip": "192.0.2.42"} response = client.put("/ban", json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert response.json['operation'] == 'add' -def test_ban_invalid(client, mocker): + +def test_ban_invalid(client, mocker, valid_credentials): def noop(): pass run_res = SimpleNamespace() @@ -52,16 +52,17 @@ def test_ban_invalid(client, mocker): mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"name": "sshd", "ip": "not:an::addr:ess"} response = client.put("/ban", json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) - assert response.json['error'] == "'not:an::addr:ess' does not appear to be an IPv4 or IPv6 address" + assert response.json['error'] == "'not:an::addr:ess' does not " \ + "appear to be an IPv4 or IPv6 address" -def test_unban_ipv6(client, mocker): +def test_unban_ipv6(client, mocker, valid_credentials): def noop(): pass run_res = SimpleNamespace() @@ -72,16 +73,16 @@ def test_unban_ipv6(client, mocker): mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"name": "sshd", "ip": "2001:db8::abad:cafe"} response = client.delete("/ban", json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert response.json['operation'] == 'delete' -def test_unban_ipv4(client, mocker): +def test_unban_ipv4(client, mocker, valid_credentials): def noop(): pass run_res = SimpleNamespace() @@ -92,10 +93,10 @@ def test_unban_ipv4(client, mocker): mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"name": "sshd", "ip": "192.0.2.42"} response = client.delete("/ban", - json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + json=json_payload, + headers={"Authorization": + "Basic " + valid_credentials}) assert response.json['operation'] == 'delete' diff --git a/tests/test_flush.py b/tests/test_flush.py index 6ea877e..9ac64ba 100644 --- a/tests/test_flush.py +++ b/tests/test_flush.py @@ -1,9 +1,8 @@ -import base64 from types import SimpleNamespace from subprocess import CalledProcessError -def test_flush(client, mocker): +def test_flush(client, mocker, valid_credentials): def noop(): pass run_res = SimpleNamespace() @@ -14,47 +13,55 @@ def test_flush(client, mocker): mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") name = 'sshd' response = client.get(f"/flush/{name}", - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert response.json['operation'] == 'flush' -def test_flush_nonexistent(client, mocker): +def test_flush_nonexistent(client, mocker, valid_credentials): - cmd = ['/usr/local/bin/sudo', '/sbin/pfctl', '-a', 'some/anchor', '-t', 'nonexistent', '-T', 'flush'] + cmd = ['/usr/local/bin/sudo', + '/sbin/pfctl', '-a', 'some/anchor', + '-t', 'nonexistent', '-T', 'flush'] + + side_effect = CalledProcessError(255, cmd, output=b'', + stderr=b'pfctl: Table does not exist') mocker.patch('jail2ban.pfctl.run', - side_effect=CalledProcessError(255, cmd, output=b'', - stderr=b'pfctl: Table does not exist')) + side_effect=side_effect) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") name = 'nonexistent' response = client.get(f"/flush/{name}", - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert 'error' in response.json -def test_wrong_method(client, mocker): +def test_wrong_method(client, mocker, valid_credentials): - cmd = ['/usr/local/bin/sudo', '/sbin/pfctl', '-a', 'some/anchor', '-t', 'nonexistent', '-T', 'flush'] + cmd = ['/usr/local/bin/sudo', + '/sbin/pfctl', '-a', 'some/anchor', + '-t', 'nonexistent', '-T', 'flush'] + + side_effect = CalledProcessError(255, cmd, output=b'', + stderr=b'pfctl: Table does not exist') mocker.patch('jail2ban.pfctl.run', - side_effect=CalledProcessError(255, cmd, output=b'', - stderr=b'pfctl: Table does not exist')) + side_effect=side_effect) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") name = 'nonexistent' response = client.put(f"/flush/{name}", - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert response.status_code == 405 -def test_filenotfound(app, mocker): +def test_filenotfound(app, mocker, valid_credentials): app.config.update({ "AUTHFILE": '../tests/nonexistent-users-test.txt' @@ -62,9 +69,9 @@ def test_filenotfound(app, mocker): client = app.test_client() - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") name = 'nonexistent' response = client.get(f"/flush/{name}", - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert response.status_code == 500 diff --git a/tests/test_register.py b/tests/test_register.py index 675ae2a..9312d09 100644 --- a/tests/test_register.py +++ b/tests/test_register.py @@ -1,13 +1,17 @@ -import base64 from subprocess import CompletedProcess - -pfctl_stdout_lines = b'''block drop quick proto tcp from to any port = submission +pfctl_stdout_lines = b''' +block drop quick proto tcp from to any port = submission block drop quick proto tcp from to any port = smtps block drop quick proto tcp from to any port = smtp block drop quick proto tcp from to any port = ssh block drop quick proto tcp from to any -''' +'''.strip() + b'\n' + +pfctl_stdout_lines_scratch = b'table persist counters\n' \ + b'block quick proto tcp from ' \ + b' to any port ' \ + b'{pop3,pop3s,imap,imaps,submission,465,sieve}\n' def test_register_unauth(client): @@ -19,7 +23,7 @@ def test_register_unauth(client): assert response.json['error'] == 'Access Denied' -def test_unregister_valid(client, mocker): +def test_unregister_valid(client, mocker, valid_credentials): def noop(): pass run_res = CompletedProcess(args=['true'], returncode=0) @@ -28,19 +32,19 @@ def test_unregister_valid(client, mocker): mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"port": "any port {pop3,pop3s,imap,imaps,submission,465,sieve}", "name": "dovecot", "protocol": "tcp"} response = client.delete("/register", json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert response.json['action'] == 'stop' -def test_register_valid(client, mocker): +def test_register_valid(client, mocker, valid_credentials): def noop(): pass run_res = CompletedProcess(args=['true'], returncode=0) @@ -49,14 +53,14 @@ def test_register_valid(client, mocker): pfctl_run = mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"port": "any port {pop3,pop3s,imap,imaps,submission,465,sieve}", "name": "dovecot", "protocol": "tcp"} response = client.put("/register", json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) pfctl_run_input_arg = pfctl_run.call_args_list[1][1]['input'] for existing_line in pfctl_stdout_lines.splitlines(): @@ -64,7 +68,8 @@ def test_register_valid(client, mocker): assert response.json['action'] == 'start' -def test_register_valid_from_scratch(client, mocker): + +def test_register_valid_from_scratch(client, mocker, valid_credentials): def noop(): pass run_res = CompletedProcess(args=['true'], returncode=0) @@ -73,22 +78,21 @@ def test_register_valid_from_scratch(client, mocker): pfctl_run = mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"port": "any port {pop3,pop3s,imap,imaps,submission,465,sieve}", "name": "dovecot", "protocol": "tcp"} response = client.put("/register", json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) pfctl_run_input_arg = pfctl_run.call_args_list[1][1]['input'] - assert pfctl_run_input_arg == b'table persist counters\nblock quick proto tcp from to any port {pop3,pop3s,imap,imaps,submission,465,sieve}\n' - + assert pfctl_run_input_arg == pfctl_stdout_lines_scratch assert response.json['action'] == 'start' -def test_register_invalid(client, mocker): +def test_register_invalid(client, mocker, valid_credentials): def noop(): pass run_res = CompletedProcess(args=['true'], returncode=0) @@ -97,13 +101,13 @@ def test_register_invalid(client, mocker): mocker.patch('jail2ban.pfctl.run', return_value=run_res) - valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8") json_payload = {"port": "not a pf statement", "name": "dovecot", "protocol": "tcp"} response = client.put("/register", json=json_payload, - headers={"Authorization": "Basic " + valid_credentials}) + headers={"Authorization": + "Basic " + valid_credentials}) assert response.json['error'] == '"not a pf statement" is tainted'