Stefan Bethke
f49cb3d6f3
All checks were successful
docker-image / docker (push) Successful in 9m31s
55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
from ipaddress import ip_address, IPv4Address, ip_network
|
|
from typing import Callable, List
|
|
|
|
from BottleOIDC import BottleOIDC
|
|
from BottleOIDC.bottle_utils import UnauthorizedError
|
|
from bottle import request
|
|
|
|
|
|
class BottleHelpers:
|
|
def __init__(self, auth: BottleOIDC, allowed=None, group=None):
|
|
if allowed is None:
|
|
self.allowed = []
|
|
else:
|
|
self.allowed = [ip_network(a) for a in allowed]
|
|
self.auth = auth
|
|
self.group = group
|
|
|
|
def require_login(self, func: Callable) -> Callable:
|
|
if self.group is not None:
|
|
return self.auth.require_login(self.auth.require_attribute('groups', self.group)(func))
|
|
else:
|
|
return self.auth.require_login(func)
|
|
|
|
def require_authz(self, func: Callable) -> Callable:
|
|
if self.group is not None:
|
|
return self.auth.require_attribute('groups', self.group)(func)
|
|
else:
|
|
def _outer_wrapper(f):
|
|
def _wrapper(*args, **kwargs):
|
|
if self.auth.my_username is not None:
|
|
return f(*args, **kwargs)
|
|
|
|
return UnauthorizedError('Not Authorized')
|
|
|
|
_wrapper.__name__ = f.__name__
|
|
return _wrapper
|
|
|
|
return _outer_wrapper(func)
|
|
|
|
def require_sourceip(self, func: Callable) -> Callable:
|
|
if self.allowed is None or len(self.allowed) == 0:
|
|
return func
|
|
|
|
def _outer_wrapper(f):
|
|
def _wrapper(*args, **kwargs):
|
|
addr = ip_network(request.remote_addr)
|
|
for allowed in self.allowed:
|
|
if addr.overlaps(allowed):
|
|
return f(*args, **kwargs)
|
|
return UnauthorizedError('Not Authorized')
|
|
|
|
_wrapper.__name__ = f.__name__
|
|
return _wrapper
|
|
|
|
return _outer_wrapper(func)
|