241 lines
7.5 KiB
Python
241 lines
7.5 KiB
Python
import json
|
|
import requests
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from collections import namedtuple
|
|
from urllib.parse import urlencode
|
|
from functools import wraps
|
|
from flask import Flask, render_template, abort, g, \
|
|
redirect, request, flash, url_for, make_response
|
|
|
|
from .dhcp import DhcpdUpdater
|
|
|
|
from spaceauth import SpaceAuth, login_required, current_user, cap_required
|
|
|
|
DeviceInfo = namedtuple('DeviceInfo', ['hwaddr', 'name', 'owner', 'ignored'])
|
|
|
|
def v4addr():
|
|
if request.headers.getlist("X-Forwarded-For"):
|
|
r_addr = request.headers.getlist("X-Forwarded-For")[-1]
|
|
else:
|
|
r_addr = request.remote_addr
|
|
if r_addr.startswith('::ffff:'):
|
|
r_addr = r_addr[7:]
|
|
return r_addr
|
|
|
|
|
|
def req_to_ctx():
|
|
return dict(iter(request.form.items()))
|
|
|
|
def get_device_info(conn, hwaddr):
|
|
return list(get_device_infos(conn, (hwaddr,)))[0]
|
|
|
|
|
|
def get_device_infos(conn, hwaddrs):
|
|
in_clause = '({})'.format(', '.join(['?'] * len(hwaddrs)))
|
|
stmt = '''select hwaddr, name, ignored, owner from
|
|
devices where devices.hwaddr in ''' + in_clause
|
|
for row in conn.execute(stmt, hwaddrs):
|
|
owner = row['owner'] or ''
|
|
ignored = row['ignored']
|
|
yield DeviceInfo(row['hwaddr'], row['name'], owner, ignored)
|
|
|
|
def app(instance_path):
|
|
app = Flask('at', instance_path=instance_path, instance_relative_config=True)
|
|
app.config.from_pyfile('at.cfg')
|
|
app.jinja_env.add_extension('jinja2.ext.i18n')
|
|
app.jinja_env.install_null_translations()
|
|
app.updater = None
|
|
|
|
if app.config.get('PROXY_FIX'):
|
|
from werkzeug.contrib.fixers import ProxyFix
|
|
app.wsgi_app = ProxyFix(app.wsgi_app)
|
|
|
|
auth = SpaceAuth(app)
|
|
|
|
def restrict_ip(prefix='', exclude=[]):
|
|
def decorator(f):
|
|
@wraps(f)
|
|
def func(*a, **kw):
|
|
r_addr = v4addr()
|
|
if not r_addr.startswith(prefix) or r_addr in exclude:
|
|
app.logger.info('got IP %s, rejecting', r_addr)
|
|
abort(403)
|
|
|
|
return f(*a, **kw)
|
|
return func
|
|
return decorator
|
|
|
|
|
|
@app.template_filter('strfts')
|
|
def strfts(ts, format='%d/%m/%Y %H:%M'):
|
|
return datetime.fromtimestamp(ts).strftime(format)
|
|
|
|
|
|
@app.template_filter('wikiurl')
|
|
def wikiurl(user):
|
|
return app.config['WIKI_URL'] % {'login': user}
|
|
|
|
|
|
@app.before_request
|
|
def make_connection():
|
|
conn = sqlite3.connect(app.config['DB'])
|
|
conn.row_factory = sqlite3.Row
|
|
conn.isolation_level = None # for autocommit mode
|
|
g.db = conn
|
|
|
|
|
|
@app.teardown_request
|
|
def close_connection(exception):
|
|
g.db.close()
|
|
|
|
|
|
@app.route('/')
|
|
def main_view():
|
|
return render_template('main.html', **now_at())
|
|
|
|
|
|
@app.route('/api')
|
|
def list_all():
|
|
data = now_at()
|
|
|
|
def prettify_user(xxx_todo_changeme):
|
|
(user, atime) = xxx_todo_changeme
|
|
if user == 'greenmaker':
|
|
user = 'dreammaker'
|
|
return {
|
|
'login': user,
|
|
'timestamp': atime,
|
|
'pretty_time': strfts(atime),
|
|
}
|
|
result = {}
|
|
result['users'] = list(map(prettify_user, data.pop('users')))
|
|
result.update((k, len(v)) for k, v in list(data.items()))
|
|
res = make_response(json.dumps(result), 200)
|
|
res.headers['Access-Control-Allow-Origin'] = '*'
|
|
return res
|
|
|
|
|
|
def now_at():
|
|
result = dict()
|
|
devices = app.updater.get_active_devices()
|
|
device_infos = list(get_device_infos(g.db, list(devices.keys())))
|
|
device_infos.sort(key=lambda di: devices[di.hwaddr])
|
|
unknown = set(devices.keys()) - set(d.hwaddr for d in device_infos)
|
|
# das kektop sorting maschine
|
|
for name, prefixes in list(app.config['SPECIAL_DEVICES'].items()):
|
|
result[name] = set()
|
|
for u in unknown.copy():
|
|
if u.startswith(prefixes):
|
|
result[name].add(u)
|
|
unknown.discard(u)
|
|
|
|
result['unknown'] = unknown
|
|
|
|
users = {}
|
|
for info in device_infos:
|
|
if info.owner not in users and not info.ignored:
|
|
users[info.owner] = devices[info.hwaddr][0]
|
|
result['users'] = sorted(list(users.items()), key=lambda u_a: u_a[1], reverse=True)
|
|
|
|
return result
|
|
|
|
|
|
restrict_to_hs = restrict_ip(prefix=app.config['CLAIMABLE_PREFIX'],
|
|
exclude=app.config['CLAIMABLE_EXCLUDE'])
|
|
|
|
|
|
@app.route('/claim', methods=['GET'])
|
|
@restrict_to_hs
|
|
@login_required
|
|
def claim_form():
|
|
hwaddr, name = app.updater.get_device(v4addr())
|
|
return render_template('claim.html', hwaddr=hwaddr, name=name)
|
|
|
|
|
|
@app.route('/claim', methods=['POST'])
|
|
@restrict_to_hs
|
|
@login_required
|
|
def claim():
|
|
hwaddr, lease_name = app.updater.get_device(v4addr())
|
|
ctx = None
|
|
if not hwaddr:
|
|
ctx = dict(error='Invalid device.')
|
|
else:
|
|
login = current_user.id
|
|
try:
|
|
g.db.execute('''
|
|
insert into devices (hwaddr, name, owner, ignored) values (?, ?, ?, ?)''',
|
|
[hwaddr, request.form['name'], login, False])
|
|
ctx = {}
|
|
except sqlite3.Error:
|
|
error = 'Could not add device! Perhaps someone claimed it?'
|
|
ctx = dict(error=error)
|
|
return render_template('post_claim.html', **ctx)
|
|
|
|
def get_user_devices(conn, user):
|
|
devs = conn.execute('select hwaddr, name, ignored from devices where\
|
|
owner = ?', [user])
|
|
device_info = []
|
|
for row in devs:
|
|
di = DeviceInfo(row['hwaddr'], row['name'], user, row['ignored'])
|
|
device_info.append(di)
|
|
return device_info
|
|
|
|
|
|
@app.route('/account', methods=['GET'])
|
|
@login_required
|
|
def account():
|
|
devices = get_user_devices(g.db, current_user.id)
|
|
return render_template('account.html', devices=devices)
|
|
|
|
|
|
def set_ignored(conn, hwaddr, user, value):
|
|
return conn.execute('''
|
|
update devices set ignored = ? where hwaddr = ? and owner = ?''',
|
|
[value, hwaddr, user])
|
|
|
|
|
|
def delete_device(conn, hwaddr, user):
|
|
return conn.execute('''
|
|
delete from devices where hwaddr = ? and owner = ?''',
|
|
[hwaddr, user])
|
|
|
|
@app.route('/devices/<id>/<action>/')
|
|
@login_required
|
|
def device(id, action):
|
|
user = current_user.id
|
|
if action == 'hide':
|
|
set_ignored(g.db, id, user, True)
|
|
if action == 'show':
|
|
set_ignored(g.db, id, user, False)
|
|
if action == 'delete':
|
|
delete_device(g.db, id, user)
|
|
return redirect(url_for('account'))
|
|
|
|
|
|
@app.route('/admin')
|
|
@cap_required('staff')
|
|
def admin():
|
|
data = now_at()
|
|
return render_template('admin.html', data=data)
|
|
|
|
|
|
#@app.before_first_request
|
|
def setup():
|
|
updater = DhcpdUpdater(app.config['LEASE_FILE'], app.config['TIMEOUT'],
|
|
app.config['LEASE_OFFSET'])
|
|
app.logger.info('parsing leases file for the first time')
|
|
updater._trigger_update()
|
|
app.logger.info('leases file parsed')
|
|
updater.start()
|
|
app.updater = updater
|
|
|
|
setup()
|
|
|
|
return app
|
|
|
|
if __name__ == '__main__':
|
|
app.logger.setLevel(logging.DEBUG)
|
|
app.run('0.0.0.0', 8080, debug=True)
|