ruben/jail2ban
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Reuse credentials as fixture / linelength fixes for readability

Browse Source
This commit is contained in:
Ruben van Staveren 2022-03-14 16:39:24 +01:00
parent 58ac46126a
commit e0d115790e
Signed by: ruben
GPG Key ID: 886F6BECD477A93F
4 changed files with 73 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
tests/conftest.py
View File

@ -1,4 +1,5 @@
import pytest import pytest
import base64
from jail2ban import create_app from jail2ban import create_app
@ -26,3 +27,8 @@ def client(app):
@pytest.fixture() @pytest.fixture()
def runner(app): def runner(app):
return app.test_cli_runner() return app.test_cli_runner()
@pytest.fixture()
def valid_credentials():
return base64.b64encode(b"test.example.com:testpassword").decode("utf-8")

35
tests/test_ban.py
View File

@ -1,8 +1,7 @@
import base64
from types import SimpleNamespace from types import SimpleNamespace
def test_ban_ipv6(client, mocker): def test_ban_ipv6(client, mocker, valid_credentials):
def noop(): def noop():
pass pass
run_res = SimpleNamespace() run_res = SimpleNamespace()
@ -13,16 +12,16 @@ def test_ban_ipv6(client, mocker):
mocker.patch('jail2ban.pfctl.run', return_value=run_res) 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"} json_payload = {"name": "sshd", "ip": "2001:db8::abad:cafe"}
response = client.put("/ban", response = client.put("/ban",
json=json_payload, json=json_payload,
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.json['operation'] == 'add' assert response.json['operation'] == 'add'
def test_ban_ipv4(client, mocker): def test_ban_ipv4(client, mocker, valid_credentials):
def noop(): def noop():
pass pass
run_res = SimpleNamespace() run_res = SimpleNamespace()
@ -33,15 +32,16 @@ def test_ban_ipv4(client, mocker):
mocker.patch('jail2ban.pfctl.run', return_value=run_res) 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"} json_payload = {"name": "sshd", "ip": "192.0.2.42"}
response = client.put("/ban", response = client.put("/ban",
json=json_payload, json=json_payload,
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.json['operation'] == 'add' assert response.json['operation'] == 'add'
def test_ban_invalid(client, mocker):
def test_ban_invalid(client, mocker, valid_credentials):
def noop(): def noop():
pass pass
run_res = SimpleNamespace() run_res = SimpleNamespace()
@ -52,16 +52,17 @@ def test_ban_invalid(client, mocker):
mocker.patch('jail2ban.pfctl.run', return_value=run_res) 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"} json_payload = {"name": "sshd", "ip": "not:an::addr:ess"}
response = client.put("/ban", response = client.put("/ban",
json=json_payload, 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(): def noop():
pass pass
run_res = SimpleNamespace() run_res = SimpleNamespace()
@ -72,16 +73,16 @@ def test_unban_ipv6(client, mocker):
mocker.patch('jail2ban.pfctl.run', return_value=run_res) 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"} json_payload = {"name": "sshd", "ip": "2001:db8::abad:cafe"}
response = client.delete("/ban", response = client.delete("/ban",
json=json_payload, json=json_payload,
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.json['operation'] == 'delete' assert response.json['operation'] == 'delete'
def test_unban_ipv4(client, mocker): def test_unban_ipv4(client, mocker, valid_credentials):
def noop(): def noop():
pass pass
run_res = SimpleNamespace() run_res = SimpleNamespace()
@ -92,10 +93,10 @@ def test_unban_ipv4(client, mocker):
mocker.patch('jail2ban.pfctl.run', return_value=run_res) 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"} json_payload = {"name": "sshd", "ip": "192.0.2.42"}
response = client.delete("/ban", response = client.delete("/ban",
json=json_payload, json=json_payload,
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.json['operation'] == 'delete' assert response.json['operation'] == 'delete'

45
tests/test_flush.py
View File

@ -1,9 +1,8 @@
import base64
from types import SimpleNamespace from types import SimpleNamespace
from subprocess import CalledProcessError from subprocess import CalledProcessError
def test_flush(client, mocker): def test_flush(client, mocker, valid_credentials):
def noop(): def noop():
pass pass
run_res = SimpleNamespace() run_res = SimpleNamespace()
@ -14,47 +13,55 @@ def test_flush(client, mocker):
mocker.patch('jail2ban.pfctl.run', return_value=run_res) mocker.patch('jail2ban.pfctl.run', return_value=run_res)
valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8")
name = 'sshd' name = 'sshd'
response = client.get(f"/flush/{name}", response = client.get(f"/flush/{name}",
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.json['operation'] == 'flush' 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', mocker.patch('jail2ban.pfctl.run',
side_effect=CalledProcessError(255, cmd, output=b'', side_effect=side_effect)
stderr=b'pfctl: Table does not exist'))
valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8")
name = 'nonexistent' name = 'nonexistent'
response = client.get(f"/flush/{name}", response = client.get(f"/flush/{name}",
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert 'error' in response.json 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', mocker.patch('jail2ban.pfctl.run',
side_effect=CalledProcessError(255, cmd, output=b'', side_effect=side_effect)
stderr=b'pfctl: Table does not exist'))
valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8")
name = 'nonexistent' name = 'nonexistent'
response = client.put(f"/flush/{name}", response = client.put(f"/flush/{name}",
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.status_code == 405 assert response.status_code == 405
def test_filenotfound(app, mocker): def test_filenotfound(app, mocker, valid_credentials):
app.config.update({ app.config.update({
"AUTHFILE": '../tests/nonexistent-users-test.txt' "AUTHFILE": '../tests/nonexistent-users-test.txt'
@ -62,9 +69,9 @@ def test_filenotfound(app, mocker):
client = app.test_client() client = app.test_client()
valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8")
name = 'nonexistent' name = 'nonexistent'
response = client.get(f"/flush/{name}", response = client.get(f"/flush/{name}",
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.status_code == 500 assert response.status_code == 500

40
tests/test_register.py
View File

@ -1,13 +1,17 @@
import base64
from subprocess import CompletedProcess from subprocess import CompletedProcess
pfctl_stdout_lines = b'''
pfctl_stdout_lines = b'''block drop quick proto tcp from <f2b-sendmail-auth> to any port = submission block drop quick proto tcp from <f2b-sendmail-auth> to any port = submission
block drop quick proto tcp from <f2b-sendmail-auth> to any port = smtps block drop quick proto tcp from <f2b-sendmail-auth> to any port = smtps
block drop quick proto tcp from <f2b-sendmail-auth> to any port = smtp block drop quick proto tcp from <f2b-sendmail-auth> to any port = smtp
block drop quick proto tcp from <f2b-sshd> to any port = ssh block drop quick proto tcp from <f2b-sshd> to any port = ssh
block drop quick proto tcp from <f2b-recidive> to any block drop quick proto tcp from <f2b-recidive> to any
''' '''.strip() + b'\n'
pfctl_stdout_lines_scratch = b'table <f2b-dovecot> persist counters\n' \
b'block quick proto tcp from <f2b-dovecot>' \
b' to any port ' \
b'{pop3,pop3s,imap,imaps,submission,465,sieve}\n'
def test_register_unauth(client): def test_register_unauth(client):
@ -19,7 +23,7 @@ def test_register_unauth(client):
assert response.json['error'] == 'Access Denied' assert response.json['error'] == 'Access Denied'
def test_unregister_valid(client, mocker): def test_unregister_valid(client, mocker, valid_credentials):
def noop(): def noop():
pass pass
run_res = CompletedProcess(args=['true'], returncode=0) 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) mocker.patch('jail2ban.pfctl.run', return_value=run_res)
valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8")
json_payload = {"port": json_payload = {"port":
"any port {pop3,pop3s,imap,imaps,submission,465,sieve}", "any port {pop3,pop3s,imap,imaps,submission,465,sieve}",
"name": "dovecot", "protocol": "tcp"} "name": "dovecot", "protocol": "tcp"}
response = client.delete("/register", response = client.delete("/register",
json=json_payload, json=json_payload,
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.json['action'] == 'stop' assert response.json['action'] == 'stop'
def test_register_valid(client, mocker): def test_register_valid(client, mocker, valid_credentials):
def noop(): def noop():
pass pass
run_res = CompletedProcess(args=['true'], returncode=0) 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) 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": json_payload = {"port":
"any port {pop3,pop3s,imap,imaps,submission,465,sieve}", "any port {pop3,pop3s,imap,imaps,submission,465,sieve}",
"name": "dovecot", "protocol": "tcp"} "name": "dovecot", "protocol": "tcp"}
response = client.put("/register", response = client.put("/register",
json=json_payload, 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'] pfctl_run_input_arg = pfctl_run.call_args_list[1][1]['input']
for existing_line in pfctl_stdout_lines.splitlines(): for existing_line in pfctl_stdout_lines.splitlines():
@ -64,7 +68,8 @@ def test_register_valid(client, mocker):
assert response.json['action'] == 'start' 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(): def noop():
pass pass
run_res = CompletedProcess(args=['true'], returncode=0) 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) 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": json_payload = {"port":
"any port {pop3,pop3s,imap,imaps,submission,465,sieve}", "any port {pop3,pop3s,imap,imaps,submission,465,sieve}",
"name": "dovecot", "protocol": "tcp"} "name": "dovecot", "protocol": "tcp"}
response = client.put("/register", response = client.put("/register",
json=json_payload, 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'] pfctl_run_input_arg = pfctl_run.call_args_list[1][1]['input']
assert pfctl_run_input_arg == b'table <f2b-dovecot> persist counters\nblock quick proto tcp from <f2b-dovecot> 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' assert response.json['action'] == 'start'
def test_register_invalid(client, mocker): def test_register_invalid(client, mocker, valid_credentials):
def noop(): def noop():
pass pass
run_res = CompletedProcess(args=['true'], returncode=0) 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) mocker.patch('jail2ban.pfctl.run', return_value=run_res)
valid_credentials = base64.b64encode(b"test.example.com:testpassword").decode("utf-8")
json_payload = {"port": json_payload = {"port":
"not a pf statement", "not a pf statement",
"name": "dovecot", "protocol": "tcp"} "name": "dovecot", "protocol": "tcp"}
response = client.put("/register", response = client.put("/register",
json=json_payload, json=json_payload,
headers={"Authorization": "Basic " + valid_credentials}) headers={"Authorization":
"Basic " + valid_credentials})
assert response.json['error'] == '"not a pf statement" is tainted' assert response.json['error'] == '"not a pf statement" is tainted'