mirror of
https://github.com/simon987/nyaa.git
synced 2025-12-14 15:49:02 +00:00
Implement range bans (#478)
* Implement range bans People connecting from banned IP ranges are unable to upload torrents anonymously, and need to manually have their accounts activated. This adds a new table "rangebans", and a command line utility, "rangeban.py", which can be used to add, list and remove rangebans from the command line. As an example: ./rangeban.py ban 192.168.0.0/24 This would rangeban anything in this /24. The temporary_tor column allows automated scripts to clean out and re-add ever-changing sets of ranges to be banned without affecting the other ranges. This has only been tested for IPv4. * Revise Rangebans Add an id column, and change "temporary_tor" to "temp". Also index masked_cidr and mask. * rangebans: fix enabled and the binary op kill me * Add enabling/disabling bans to rangeban.py * rangebans: fail earlier on garbage arguments * rangebans: fix linter errors * rangeban.py: don't shadow builtin keyword 'id' * rangebans: change temporary ban logic, column The 'temp' column is now a nullable time column. If the field is null, the ban is understood to be permanent. If there is a time in there, it's understood to be the creation time of the ban. This allows scripts to e.g. delete all temporary bans older than a certain amount of time. Also, rename the '_cidr_string' column to 'cidr_string', because reasons. * rangeban.py: use ip_address to parse CIDR subnet * rangebans: fixes to the mask calculation and query Both were not bugs per-se, but just technically not needed/correct. * De-meme apparently
This commit is contained in:
119
rangeban.py
Executable file
119
rangeban.py
Executable file
@@ -0,0 +1,119 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from datetime import datetime
|
||||
from ipaddress import ip_address
|
||||
import sys
|
||||
|
||||
import click
|
||||
|
||||
from nyaa import create_app, models
|
||||
from nyaa.extensions import db
|
||||
|
||||
|
||||
def is_cidr_valid(c):
|
||||
'''Checks whether a CIDR range string is valid.'''
|
||||
try:
|
||||
subnet, mask = c.split('/')
|
||||
except ValueError:
|
||||
return False
|
||||
if int(mask) < 1 or int(mask) > 32:
|
||||
return False
|
||||
try:
|
||||
ip = ip_address(subnet)
|
||||
except ValueError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def check_str(b):
|
||||
'''Returns a checkmark or cross depending on the condition.'''
|
||||
return '\u2713' if b else '\u2717'
|
||||
|
||||
|
||||
@click.group()
|
||||
def rangeban():
|
||||
global app
|
||||
app = create_app('config')
|
||||
|
||||
|
||||
@rangeban.command()
|
||||
@click.option('--temp/--no-temp', help='Mark this entry as one that may be '
|
||||
'cleaned out occasionally.', default=False)
|
||||
@click.argument('cidrrange')
|
||||
def ban(temp, cidrrange):
|
||||
if not is_cidr_valid(cidrrange):
|
||||
click.secho('{} is not of the format xxx.xxx.xxx.xxx/xx.'
|
||||
.format(cidrrange), err=True, fg='red')
|
||||
sys.exit(1)
|
||||
with app.app_context():
|
||||
ban = models.RangeBan(cidr_string=cidrrange, temp=datetime.utcnow() if temp else None)
|
||||
db.session.add(ban)
|
||||
db.session.commit()
|
||||
click.echo('Added {} for {}.'.format('temp ban' if temp else 'ban',
|
||||
cidrrange))
|
||||
|
||||
|
||||
@rangeban.command()
|
||||
@click.argument('cidrrange')
|
||||
def unban(cidrrange):
|
||||
if not is_cidr_valid(cidrrange):
|
||||
click.secho('{} is not of the format xxx.xxx.xxx.xxx/xx.'
|
||||
.format(cidrrange), err=True, fg='red')
|
||||
sys.exit(1)
|
||||
with app.app_context():
|
||||
# Dunno why this wants _cidr_string and not cidr_string, probably
|
||||
# due to this all being a janky piece of shit.
|
||||
bans = models.RangeBan.query.filter(
|
||||
models.RangeBan._cidr_string == cidrrange).all()
|
||||
if len(bans) == 0:
|
||||
click.echo('Ban not found.')
|
||||
for b in bans:
|
||||
click.echo('Unbanned {}'.format(b.cidr_string))
|
||||
db.session.delete(b)
|
||||
db.session.commit()
|
||||
|
||||
|
||||
@rangeban.command()
|
||||
def list():
|
||||
with app.app_context():
|
||||
bans = models.RangeBan.query.all()
|
||||
if len(bans) == 0:
|
||||
click.echo('No bans.')
|
||||
else:
|
||||
click.secho('ID CIDR Range Enabled Temp', bold=True)
|
||||
for b in bans:
|
||||
click.echo('{0: <6} {1: <18} {2: <7} {3: <4}'
|
||||
.format(b.id, b.cidr_string,
|
||||
check_str(b.enabled),
|
||||
check_str(b.temp is not None)))
|
||||
|
||||
@rangeban.command()
|
||||
@click.argument('banid', type=int)
|
||||
@click.argument('status')
|
||||
def enabled(banid, status):
|
||||
yeses = ['true', '1', 'yes', '\u2713']
|
||||
noses = ['false', '0', 'no', '\u2717']
|
||||
if status.lower() in yeses:
|
||||
set_to = True
|
||||
elif status.lower() in noses:
|
||||
set_to = False
|
||||
else:
|
||||
click.secho('Please choose one of {} or {}.'
|
||||
.format(yeses, noses), err=True, fg='red')
|
||||
sys.exit(1)
|
||||
with app.app_context():
|
||||
ban = models.RangeBan.query.get(banid)
|
||||
if not ban:
|
||||
click.secho('No ban with id {} found.'
|
||||
.format(banid), err=True, fg='red')
|
||||
sys.exit(1)
|
||||
ban.enabled = set_to
|
||||
db.session.add(ban)
|
||||
db.session.commit()
|
||||
click.echo('{} ban {} on {}.'.format('Enabled' if set_to else 'Disabled',
|
||||
banid, ban._cidr_string))
|
||||
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
rangeban()
|
||||
Reference in New Issue
Block a user