lint fixes

This commit is contained in:
Ruben van Staveren 2024-07-31 16:14:52 +02:00
parent 0b1294b6e7
commit 63c8e39b59
Signed by: ruben
GPG Key ID: 886F6BECD477A93F
4 changed files with 51 additions and 18 deletions

View File

@ -1,6 +1,7 @@
[![pipeline status](https://gitlab.niet.verweg.com/ruben/jail2ban-pf/badges/main/pipeline.svg)](https://gitlab.niet.verweg.com/ruben/jail2ban-pf/-/commits/main) [![pipeline status](https://gitlab.niet.verweg.com/ruben/jail2ban-pf/badges/main/pipeline.svg)](https://gitlab.niet.verweg.com/ruben/jail2ban-pf/-/commits/main)
[![coverage report](https://gitlab.niet.verweg.com/ruben/jail2ban-pf/badges/main/coverage.svg)](https://gitlab.niet.verweg.com/ruben/jail2ban-pf/-/commits/main) [![coverage report](https://gitlab.niet.verweg.com/ruben/jail2ban-pf/badges/main/coverage.svg)](https://gitlab.niet.verweg.com/ruben/jail2ban-pf/-/commits/main)
An API to remotely control a pf based fail2ban
## Installation ## Installation

View File

@ -1,12 +1,18 @@
from flask import Flask, request, jsonify, current_app '''
from flask_httpauth import HTTPBasicAuth An API to remotely control a pf based fail2ban
from werkzeug.security import check_password_hash '''
from ipaddress import ip_address
import re import re
from jail2ban.pfctl import pfctl_table_op, pfctl_cfg_read, pfctl_cfg_write from ipaddress import ip_address
from jail2ban.auth import get_users
from subprocess import CalledProcessError from subprocess import CalledProcessError
from flask import Flask, current_app, jsonify, request
from flask_httpauth import HTTPBasicAuth # type: ignore
from werkzeug.security import check_password_hash
from jail2ban.auth import get_users
from jail2ban.pfctl import pfctl_cfg_read, pfctl_cfg_write, pfctl_table_op
auth = HTTPBasicAuth() auth = HTTPBasicAuth()
@ -16,48 +22,52 @@ PAT_PROT = r'^(?:tcp|udp)$'
PAT_NAME = r'^[\w\-]+$' PAT_NAME = r'^[\w\-]+$'
def untaint(pattern, string): def untaint(pattern: str, string: str) -> str:
''' '''
untaint string (as perl does) untaint string (as perl does)
''' '''
match = re.match(pattern, string) match = re.match(pattern, string)
if match: if match:
return match.string return match.string
else:
raise ValueError(f'"{string}" is tainted') raise ValueError(f'"{string}" is tainted')
def create_app(): def create_app():
'''
Create the flask application
'''
app = Flask(__name__, instance_relative_config=True) app = Flask(__name__, instance_relative_config=True)
# load the instance config, if it exists, when not testing # load the instance config, if it exists, when not testing
app.config.from_pyfile('config.py', silent=True) app.config.from_pyfile('config.py', silent=True)
@auth.verify_password @auth.verify_password
def verify_password(username, password): def verify_password(username: str, password: str) -> str | None:
users = get_users() users = get_users()
current_app.logger.debug(users) current_app.logger.debug(users)
current_app.logger.debug('Checking password of %s', username) current_app.logger.debug('Checking password of %s', username)
if username in users and \ if username in users and \
check_password_hash(users.get(username), password): check_password_hash(users.get(username), password):
return username return username
return None
@app.route("/ping", methods=['GET']) @app.route("/ping", methods=['GET'])
@auth.login_required @auth.login_required
def ping(): def ping():
remote_user = auth.username() remote_user = auth.username()
app.logger.info('Received ping for' app.logger.info('Received ping for'
f' anchor f2b-jail/{remote_user}') ' anchor f2b-jail/%s', remote_user)
return jsonify({'anchor': f'f2b-jail/{remote_user}', return jsonify({'anchor': f'f2b-jail/{remote_user}',
'operation': 'ping', 'operation': 'ping',
'result': 'pong'}) 'result': 'pong'})
@app.route("/flush/<name>", methods=['GET']) @app.route("/flush/<name>", methods=['GET'])
@auth.login_required @auth.login_required
def flush(name): def flush(name):
remote_user = auth.username() remote_user = auth.username()
name = untaint(PAT_NAME, name) name = untaint(PAT_NAME, name)
app.logger.info(f'Flushing table f2b-{name}' app.logger.info('Flushing table f2b-%s'
f' in anchor f2b-jail/{remote_user}') ' in anchor f2b-jail/%s', name, remote_user)
res = pfctl_table_op('f2b-jail/{remote_user}', res = pfctl_table_op('f2b-jail/{remote_user}',
table='f2b-{name}', table='f2b-{name}',
operation='flush') operation='flush')
@ -97,7 +107,7 @@ def create_app():
pfctl_table_op(f'f2b-jail/{remote_user}', pfctl_table_op(f'f2b-jail/{remote_user}',
table=f'f2b-{name}', table=f'f2b-{name}',
operation='kill') operation='kill')
app.logger.info(f'pfctl -a f2b-jail/{remote_user} -f-') app.logger.info('pfctl -a f2b-jail/%s -f-', remote_user)
return jsonify({'anchor': f'f2b-jail/{remote_user}', return jsonify({'anchor': f'f2b-jail/{remote_user}',
'table': f'f2b-{name}', 'table': f'f2b-{name}',
'action': 'start' if request.method == 'PUT' 'action': 'start' if request.method == 'PUT'
@ -113,15 +123,15 @@ def create_app():
name = untaint(PAT_NAME, data['name']) name = untaint(PAT_NAME, data['name'])
ip = ip_address(data['ip']) ip = ip_address(data['ip'])
if request.method == 'PUT': if request.method == 'PUT':
app.logger.info(f'Add {ip} to f2b-{name}' app.logger.info('Add %s to f2b-%s'
f' in anchor f2b-jail/{remote_user}') ' in anchor f2b-jail/%s', ip, name, remote_user)
res = pfctl_table_op(f'f2b-jail/{remote_user}', res = pfctl_table_op(f'f2b-jail/{remote_user}',
table=f'f2b-{name}', table=f'f2b-{name}',
operation='add', operation='add',
value=str(ip)) value=str(ip))
else: # 'DELETE': else: # 'DELETE':
app.logger.info(f'Remove {ip} from f2b-{name}' app.logger.info('Remove %s from f2b-%s'
f' in anchor f2b-jail/{remote_user}') ' in anchor f2b-jail/%s', ip, name, remote_user)
res = pfctl_table_op(f'f2b-jail/{remote_user}', res = pfctl_table_op(f'f2b-jail/{remote_user}',
table=f'f2b-{name}', table=f'f2b-{name}',
operation='delete', operation='delete',

View File

@ -1,7 +1,13 @@
'''
Authentication backend
'''
from flask import current_app from flask import current_app
def get_users(): def get_users():
'''
Load users from password file (AUTHFILE)
'''
users = {} users = {}
authfile = current_app.config['AUTHFILE'] authfile = current_app.config['AUTHFILE']

View File

@ -1,3 +1,6 @@
'''
pf table/anchor operations
'''
import logging import logging
from subprocess import run from subprocess import run
@ -6,6 +9,9 @@ _PFCTL = '/sbin/pfctl'
def pfctl_cfg_read(anchor): def pfctl_cfg_read(anchor):
'''
Read from pf anchor
'''
cmd = [_SUDO, _PFCTL, '-a', anchor, '-sr'] cmd = [_SUDO, _PFCTL, '-a', anchor, '-sr']
logging.info('Running %s', cmd) logging.info('Running %s', cmd)
@ -16,6 +22,9 @@ def pfctl_cfg_read(anchor):
def pfctl_cfg_write(anchor, cfg): def pfctl_cfg_write(anchor, cfg):
'''
Apply configuration to pf anchor
'''
cmd = [_SUDO, _PFCTL, '-a', anchor, '-f-'] cmd = [_SUDO, _PFCTL, '-a', anchor, '-f-']
logging.info('Running %s', cmd) logging.info('Running %s', cmd)
logging.info('Config %s', cfg) logging.info('Config %s', cfg)
@ -30,6 +39,13 @@ def pfctl_cfg_write(anchor, cfg):
def pfctl_table_op(anchor, **kwargs): def pfctl_table_op(anchor, **kwargs):
'''
pf table operation
Parameters:
* table: which table to work on
* operation: operation used on the table
* value (optional): value used for the operation
'''
table = kwargs['table'] table = kwargs['table']
operation = kwargs['operation'] operation = kwargs['operation']
value = kwargs['value'] if 'value' in kwargs else None value = kwargs['value'] if 'value' in kwargs else None