Compare commits

..

10 Commits

19 changed files with 559 additions and 404 deletions

View File

@ -12,6 +12,7 @@ from . import logger
from .plugins import AuthPlugin
from flaschengeist.config import config, configure_app
from flaschengeist.controller import roleController
from flaschengeist.utils.hook import Hook
class CustomJSONEncoder(JSONEncoder):
@ -35,6 +36,7 @@ class CustomJSONEncoder(JSONEncoder):
return JSONEncoder.default(self, o)
@Hook("plugins.loaded")
def __load_plugins(app):
logger.debug("Search for plugins")
@ -70,31 +72,30 @@ def __load_plugins(app):
else:
logger.info(f"Using plugin: {entry_point.name}")
app.config["FG_PLUGINS"][entry_point.name] = plugin
else:
logger.debug(f"Skip disabled plugin {entry_point.name}")
if "FG_AUTH_BACKEND" not in app.config:
logger.error("No authentication plugin configured or authentication plugin not found")
raise RuntimeError("No authentication plugin configured or authentication plugin not found")
@Hook("plugins.installed")
def install_all():
from flaschengeist.database import db
db.create_all()
db.session.commit()
installed = []
for name, plugin in current_app.config["FG_PLUGINS"].items():
if not plugin:
logger.debug(f"Skip disabled plugin: {name}")
continue
logger.info(f"Install plugin {name}")
plugin.install()
installed.append(plugin)
if plugin.permissions:
roleController.create_permissions(plugin.permissions)
for plugin in installed:
plugin.post_install()
def create_app(test_config=None):
def create_app(test_config=None, cli=False):
app = Flask(__name__)
app.json_encoder = CustomJSONEncoder
CORS(app)
@ -102,7 +103,7 @@ def create_app(test_config=None):
with app.app_context():
from flaschengeist.database import db
configure_app(app, test_config)
configure_app(app, test_config, cli)
db.init_app(app)
__load_plugins(app)
@ -112,14 +113,6 @@ def create_app(test_config=None):
return jsonify({"plugins": app.config["FG_PLUGINS"], "version": version})
@app.after_request
def after_request(response):
header = response.headers
header["Access-Control-Allow-Origin"] = "*"
header["Access-Control-Allow-Methods"] = "GET,HEAD,OPTIONS,POST,PUT"
header["Access-Control-Allow-Headers"] = "Origin, X-Requested-With, Content-Type, Accept, Authorization"
return response
@app.errorhandler(Exception)
def handle_exception(e):
if isinstance(e, HTTPException):

View File

@ -0,0 +1,107 @@
import sys
import inspect
import logging
class InterfaceGenerator:
known = []
classes = {}
mapper = {
"str": "string",
"int": "number",
"float": "number",
"date": "Date",
"datetime": "Date",
"NoneType": "null",
"bool": "boolean",
}
def __init__(self, namespace, filename, logger=logging.getLogger()):
self.basename = ""
self.namespace = namespace
self.filename = filename
self.this_type = None
self.logger = logger
def pytype(self, cls):
a = self._pytype(cls)
return a
def _pytype(self, cls):
import typing
origin = typing.get_origin(cls)
arguments = typing.get_args(cls)
if origin is typing.ForwardRef: # isinstance(cls, typing.ForwardRef):
return "", "this" if cls.__forward_arg__ == self.this_type else cls.__forward_arg__
if origin is typing.Union:
if len(arguments) == 2 and arguments[1] is type(None):
return "?", self.pytype(arguments[0])[1]
else:
return "", "|".join([self.pytype(pt)[1] for pt in arguments])
if origin is list:
return "", "Array<{}>".format("|".join([self.pytype(a_type)[1] for a_type in arguments]))
if cls is typing.Any:
return "", "any"
name = cls.__name__ if hasattr(cls, "__name__") else cls if isinstance(cls, str) else None
if name is not None:
if name in self.mapper:
return "", self.mapper[name]
else:
return "", name
self.logger.warning(f"This python version might not detect all types (try >= 3.9). Could not identify >{cls}<")
return "?", "any"
def walker(self, module):
if sys.version_info < (3, 9):
raise RuntimeError("Python >= 3.9 is required to export API")
import typing
if (
inspect.ismodule(module[1])
and module[1].__name__.startswith(self.basename)
and module[1].__name__ not in self.known
):
self.known.append(module[1].__name__)
for cls in inspect.getmembers(module[1], lambda x: inspect.isclass(x) or inspect.ismodule(x)):
self.walker(cls)
elif (
inspect.isclass(module[1])
and module[1].__module__.startswith(self.basename)
and module[0] not in self.classes
and not module[0].startswith("_")
and hasattr(module[1], "__annotations__")
):
self.this_type = module[0]
d = {}
for param, ptype in typing.get_type_hints(module[1], globalns=None, localns=None).items():
if not param.startswith("_") and not param.endswith("_"):
d[param] = self.pytype(ptype)
if len(d) == 1:
key, value = d.popitem()
self.classes[module[0]] = value[1]
else:
self.classes[module[0]] = d
def run(self, models):
self.basename = models.__name__
self.walker(("models", models))
def write(self):
with (open(self.filename, "w") if self.filename else sys.stdout) as file:
file.write("declare namespace {} {{\n".format(self.namespace))
for cls, params in self.classes.items():
if isinstance(params, str):
file.write("\ttype {} = {};\n".format(cls, params))
else:
file.write("\tinterface {} {{\n".format(cls))
for name in params:
file.write("\t\t{}{}: {};\n".format(name, *params[name]))
file.write("\t}\n")
file.write("}\n")

View File

@ -0,0 +1,160 @@
import pathlib
import subprocess
import click
from os import environ
from flask import current_app
from flask.cli import FlaskGroup, run_command, with_appcontext
import pkg_resources
from flaschengeist.app import create_app
from flaschengeist.config import configure_logger
def get_version(ctx, param, value):
if not value or ctx.resilient_parsing:
return
import platform
from werkzeug import __version__ as werkzeug_version
from flask import __version__ as flask_version
from flaschengeist import __version__
click.echo(
f"Python {platform.python_version()}\n"
f"Flask {flask_version}\n"
f"Werkzeug {werkzeug_version}\n"
f"Flaschengeist {__version__}",
color=ctx.color,
)
ctx.exit()
@with_appcontext
def verbosity(ctx, param, value):
"""Toggle verbosity between WARNING <-> DEBUG"""
if not value or ctx.resilient_parsing:
return
configure_logger(cli=40 - max(0, min(value * 10, 30)))
@click.group(
cls=FlaskGroup,
add_version_option=False,
add_default_commands=False,
create_app=lambda: create_app(cli=True),
)
@click.option(
"--version",
help="Show the flask version",
expose_value=False,
callback=get_version,
is_flag=True,
is_eager=True,
)
@click.option(
"--verbose",
"-v",
help="Increase logging level",
callback=verbosity,
count=True,
expose_value=False,
)
def cli():
"""Management script for the Flaschengeist application."""
pass
@cli.command()
@with_appcontext
def install():
"""Install and initialize enabled plugins.
Most plugins need to install custom tables into the database
running this command will lookup all enabled plugins and run
their database initalization routines.
"""
from flaschengeist.app import install_all
install_all()
@cli.command()
@click.option("--output", "-o", help="Output file, default is stdout", type=click.Path())
@click.option("--namespace", "-n", help="TS namespace for the interfaces", type=str, show_default=True)
@click.option("--plugin", "-p", help="Also export types for a plugin (even if disabled)", multiple=True, type=str)
@click.option("--no-core", help="Skip models / types from flaschengeist core", is_flag=True)
def export(namespace, output, no_core, plugin):
from flaschengeist import models
from flaschengeist import logger
from .InterfaceGenerator import InterfaceGenerator
gen = InterfaceGenerator(namespace, output, logger)
if not no_core:
gen.run(models)
if plugin:
for entry_point in pkg_resources.iter_entry_points("flaschengeist.plugins"):
if len(plugin) == 0 or entry_point.name in plugin:
plg = entry_point.load()
if hasattr(plg, "models") and plg.models is not None:
gen.run(plg.models)
gen.write()
@cli.command()
@click.option("--host", help="set hostname to listen on", default="127.0.0.1", show_default=True)
@click.option("--port", help="set port to listen on", type=int, default=5000, show_default=True)
@click.option("--debug", help="run in debug mode", is_flag=True)
@with_appcontext
@click.pass_context
def run(ctx, host, port, debug):
"""Run Flaschengeist using a development server."""
class PrefixMiddleware(object):
def __init__(self, app, prefix=""):
self.app = app
self.prefix = prefix
def __call__(self, environ, start_response):
if environ["PATH_INFO"].startswith(self.prefix):
environ["PATH_INFO"] = environ["PATH_INFO"][len(self.prefix) :]
environ["SCRIPT_NAME"] = self.prefix
return self.app(environ, start_response)
else:
start_response("404", [("Content-Type", "text/plain")])
return ["This url does not belong to the app.".encode()]
from flaschengeist.config import config
# re configure logger, as we are no logger in CLI mode
configure_logger()
current_app.wsgi_app = PrefixMiddleware(current_app.wsgi_app, prefix=config["FLASCHENGEIST"].get("root", ""))
if debug:
environ["FLASK_DEBUG"] = "1"
environ["FLASK_ENV"] = "development"
ctx.invoke(run_command, host=host, port=port, debugger=debug)
@cli.command()
@click.option(
"--output",
"-o",
help="Documentation output path",
default="./docs",
type=click.Path(file_okay=False, path_type=pathlib.Path),
)
def docs(output: pathlib.Path):
"""Generate and export API documentation using pdoc3"""
output.mkdir(parents=True, exist_ok=True)
command = [
"python",
"-m",
"pdoc",
"--skip-errors",
"--html",
"--output-dir",
str(output),
"flaschengeist",
]
click.echo(f"Running command: {command}")
subprocess.check_call(command)

View File

@ -33,7 +33,7 @@ def read_configuration(test_config):
for loc in paths:
try:
with (loc / "flaschengeist.toml").open() as source:
print("Reading config file from >{}<".format(loc))
logger.warning(f"Reading config file from >{loc}<") # default root logger, goes to stderr
update_dict(config, toml.load(source))
except IOError:
pass
@ -41,7 +41,7 @@ def read_configuration(test_config):
update_dict(config, test_config)
def configure_logger():
def configure_logger(cli=False):
global config
# Read default config
logger_config = toml.load(_module_path / "logging.toml")
@ -50,25 +50,27 @@ def configure_logger():
# Override with user config
update_dict(logger_config, config.get("LOGGING"))
# Check for shortcuts
if "level" in config["LOGGING"]:
logger_config["loggers"]["flaschengeist"] = {"level": config["LOGGING"]["level"]}
logger_config["handlers"]["console"]["level"] = config["LOGGING"]["level"]
logger_config["handlers"]["file"]["level"] = config["LOGGING"]["level"]
if not config["LOGGING"].get("console", True):
if "level" in config["LOGGING"] or isinstance(cli, int):
level = cli if isinstance(cli, int) else config["LOGGING"]["level"]
logger_config["loggers"]["flaschengeist"] = {"level": level}
logger_config["handlers"]["console"]["level"] = level
logger_config["handlers"]["file"]["level"] = level
if cli is True or not config["LOGGING"].get("console", True):
logger_config["handlers"]["console"]["level"] = "CRITICAL"
if "file" in config["LOGGING"]:
if not cli and isinstance(config["LOGGING"].get("file", False), str):
logger_config["root"]["handlers"].append("file")
logger_config["handlers"]["file"]["filename"] = config["LOGGING"]["file"]
path = Path(config["LOGGING"]["file"])
path.parent.mkdir(parents=True, exist_ok=True)
Path(config["LOGGING"]["file"]).parent.mkdir(parents=True, exist_ok=True)
else:
del logger_config["handlers"]["file"]
logging.config.dictConfig(logger_config)
def configure_app(app, test_config=None):
def configure_app(app, test_config=None, cli=False):
global config
read_configuration(test_config)
configure_logger()
configure_logger(cli)
# Always enable this builtin plugins!
update_dict(

View File

@ -17,7 +17,7 @@ def get(role_name) -> Role:
else:
role = Role.query.filter(Role.name == role_name).one_or_none()
if not role:
raise NotFound
raise NotFound("no such role")
return role
@ -56,11 +56,14 @@ def create_permissions(permissions):
def create_role(name: str, permissions=[]):
logger.debug(f"Create new role with name: {name}")
try:
role = Role(name=name)
db.session.add(role)
set_permissions(role, permissions)
db.session.commit()
logger.debug(f"Created role: {role.serialize()}")
except IntegrityError:
raise BadRequest("role already exists")
return role

View File

@ -1,4 +1,5 @@
import secrets
import re
from io import BytesIO
from sqlalchemy import exc
from flask import current_app
@ -90,15 +91,13 @@ def update_user(user):
db.session.commit()
def set_roles(user: User, roles: list[str], create=False):
user.roles_.clear()
for role_name in roles:
role = Role.query.filter(Role.name == role_name).one_or_none()
if not role:
if not create:
raise BadRequest("Role not found >{}<".format(role_name))
role = Role(name=role_name)
user.roles_.append(role)
def set_roles(user: User, roles: list[str]):
if not isinstance(roles, list) and any([not isinstance(r, str) for r in roles]):
raise BadRequest("Invalid role name")
fetched = Role.query.filter(Role.name.in_(roles)).all()
if len(fetched) < len(roles):
raise BadRequest("Invalid role name, role not found")
user.roles_ = fetched
def modify_user(user, password, new_password=None):
@ -216,20 +215,31 @@ def delete_user(user: User):
db.session.commit()
def register(data):
def register(data, passwd=None):
"""Register a new user
Args:
data: dictionary containing valid user properties
passwd: optional a password, default: 16byte random
"""
allowed_keys = User().serialize().keys()
values = {key: value for key, value in data.items() if key in allowed_keys}
roles = values.pop("roles", [])
if "birthday" in data:
values["birthday"] = from_iso_format(data["birthday"]).date()
if "mail" in data and not re.match(r"[^@]+@[^@]+\.[^@]+", data["mail"]):
raise BadRequest("Invalid mail given")
user = User(**values)
set_roles(user, roles)
password = secrets.token_urlsafe(16)
password = passwd if passwd else secrets.token_urlsafe(16)
current_app.config["FG_AUTH_BACKEND"].create_user(user, password)
try:
db.session.add(user)
db.session.commit()
except exc.IntegrityError:
raise BadRequest("userid already in use")
if user.mail:
reset = _generate_password_reset(user)
subject = str(config["MESSAGES"]["welcome_subject"]).format(name=user.display_name, username=user.userid)

View File

@ -1,9 +1,31 @@
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import MetaData
db = SQLAlchemy()
# https://alembic.sqlalchemy.org/en/latest/naming.html
metadata = MetaData(
naming_convention={
"pk": "pk_%(table_name)s",
"ix": "ix_%(table_name)s_%(column_0_name)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
}
)
db = SQLAlchemy(metadata=metadata)
def case_sensitive(s):
"""
Compare string as case sensitive on the database
Args:
s: string to compare
Example:
User.query.filter(User.name == case_sensitive(some_string))
"""
if db.session.bind.dialect.name == "mysql":
from sqlalchemy import func

View File

@ -23,10 +23,11 @@ secret_key = "V3ryS3cr3t"
#
# Logging level, possible: DEBUG INFO WARNING ERROR
level = "DEBUG"
# Uncomment to enable logging to a file
# Logging to a file is simple, just add the path
# file = "/tmp/flaschengeist-debug.log"
file = false
# Uncomment to disable console logging
# console = False
# console = false
[DATABASE]
# engine = "mysql" (default)

View File

@ -15,7 +15,7 @@ disable_existing_loggers = false
class = "logging.StreamHandler"
level = "DEBUG"
formatter = "simple"
stream = "ext://sys.stdout"
stream = "ext://sys.stderr"
[handlers.file]
class = "logging.handlers.WatchedFileHandler"
level = "WARNING"

View File

@ -53,7 +53,7 @@ class User(db.Model, ModelSerializeMixin):
"""
__tablename__ = "user"
userid: str = db.Column(db.String(30), nullable=False)
userid: str = db.Column(db.String(30), unique=True, nullable=False)
display_name: str = db.Column(db.String(30))
firstname: str = db.Column(db.String(50), nullable=False)
lastname: str = db.Column(db.String(50), nullable=False)

View File

@ -10,6 +10,21 @@ from flaschengeist.models.user import _Avatar, User
from flaschengeist.models.setting import _PluginSetting
from flaschengeist.utils.hook import HookBefore, HookAfter
plugins_installed = HookAfter("plugins.installed")
"""Hook decorator for when all plugins are installed
Possible use case would be to populate the database with some presets.
Args:
hook_result: void (kwargs)
"""
plugins_loaded = HookAfter("plugins.loaded")
"""Hook decorator for when all plugins are loaded
Possible use case would be to check if a specific other plugin is loaded and change own behavior
Args:
app: Current flask app instance (args)
hook_result: void (kwargs)
"""
before_role_updated = HookBefore("update_role")
"""Hook decorator for when roles are modified
Args:
@ -57,12 +72,6 @@ class Plugin:
"""
pass
def post_install(self):
"""Fill database or do other stuff
Called after all plugins are installed
"""
pass
def get_setting(self, name: str, **kwargs):
"""Get plugin setting from database
Args:

View File

@ -0,0 +1,27 @@
import click
from flask import current_app
from flask.cli import with_appcontext
@click.command(no_args_is_help=True)
@click.option("--sync", is_flag=True, default=False, help="Synchronize users from LDAP -> database")
@with_appcontext
@click.pass_context
def ldap(ctx, sync):
"""Tools for the LDAP authentification"""
if sync:
from flaschengeist.controller import userController
from flaschengeist.plugins.auth_ldap import AuthLDAP
from ldap3 import SUBTREE
auth_ldap: AuthLDAP = current_app.config.get("FG_AUTH_BACKEND")
if auth_ldap is None or not isinstance(auth_ldap, AuthLDAP):
ctx.fail("auth_ldap plugin not found or not enabled!")
conn = auth_ldap.ldap.connection
if not conn:
conn = auth_ldap.ldap.connect(auth_ldap.root_dn, auth_ldap.root_secret)
conn.search(auth_ldap.search_dn, "(uid=*)", SUBTREE, attributes=["uid", "givenName", "sn", "mail"])
ldap_users_response = conn.response
for ldap_user in ldap_users_response:
uid = ldap_user["attributes"]["uid"][0]
userController.find_user(uid)

View File

@ -7,13 +7,16 @@ import os
import hashlib
import binascii
from werkzeug.exceptions import BadRequest
from flaschengeist.plugins import AuthPlugin
from flaschengeist.plugins import AuthPlugin, plugins_installed
from flaschengeist.models.user import User, Role, Permission
from flaschengeist.database import db
from flaschengeist import logger
class AuthPlain(AuthPlugin):
def install(self):
plugins_installed(self.post_install)
def post_install(self):
if User.query.filter(User.deleted == False).count() == 0:
logger.info("Installing admin user")

View File

@ -0,0 +1,50 @@
import click
from flask.cli import with_appcontext
from werkzeug.exceptions import BadRequest, Conflict, NotFound
from flaschengeist.controller import roleController, userController
USER_KEY = f"{__name__}.user"
def user(ctx, param, value):
if not value or ctx.resilient_parsing:
return
click.echo("Adding new user")
ctx.meta[USER_KEY] = {}
try:
ctx.meta[USER_KEY]["userid"] = click.prompt("userid", type=str)
ctx.meta[USER_KEY]["firstname"] = click.prompt("firstname", type=str)
ctx.meta[USER_KEY]["lastname"] = click.prompt("lastname", type=str)
ctx.meta[USER_KEY]["display_name"] = click.prompt("displayed name", type=str, default="")
ctx.meta[USER_KEY]["mail"] = click.prompt("mail", type=str, default="")
ctx.meta[USER_KEY]["password"] = click.prompt("password", type=str, confirmation_prompt=True, hide_input=True)
ctx.meta[USER_KEY] = {k: v for k, v in ctx.meta[USER_KEY].items() if v != ""}
except click.Abort:
click.echo("\n!!! User was not added, aborted.")
del ctx.meta[USER_KEY]
@click.command()
@click.option("--add-role", help="Add new role", type=str)
@click.option("--set-admin", help="Make a role an admin role, adding all permissions", type=str)
@click.option("--add-user", help="Add new user interactivly", callback=user, is_flag=True, expose_value=False)
@with_appcontext
def users(add_role, set_admin):
from flaschengeist.database import db
ctx = click.get_current_context()
try:
if add_role:
roleController.create_role(add_role)
if set_admin:
role = roleController.get(set_admin)
role.permissions = roleController.get_permissions()
db.session.commit()
if USER_KEY in ctx.meta:
userController.register(ctx.meta[USER_KEY], ctx.meta[USER_KEY]["password"])
except (BadRequest, NotFound) as e:
ctx.fail(e.description)

View File

@ -1,43 +1,69 @@
_hook_dict = ({}, {})
from functools import wraps
class Hook(object):
"""Decorator for Hooks
Use to decorate system hooks where plugins should be able to hook in
_hooks_before = {}
_hooks_after = {}
def Hook(function=None, id=None):
"""Hook decorator
Use to decorate functions as hooks, so plugins can hook up their custom functions.
"""
# `id` passed as `arg` not `kwarg`
if isinstance(function, str):
return Hook(id=function)
def decorator(function):
@wraps(function)
def wrapped(*args, **kwargs):
_id = id if id is not None else function.__qualname__
# Hooks before
for f in _hooks_before.get(_id, []):
f(*args, **kwargs)
# Main function
result = function(*args, **kwargs)
# Hooks after
for f in _hooks_after.get(_id, []):
f(*args, hook_result=result, **kwargs)
return result
return wrapped
# Called @Hook or we are in the second step
if callable(function):
return decorator(function)
else:
return decorator
def HookBefore(id: str):
"""Decorator for functions to be called before a Hook-Function is called
The hooked up function must accept the same arguments as the function hooked onto,
as the functions are called with the same arguments.
Hint: This enables you to modify the arguments!
"""
if not id or not isinstance(id, str):
raise TypeError("HookBefore requires the ID of the function to hook up")
def wrapped(function):
_hooks_before.setdefault(id, []).append(function)
return function
return wrapped
def HookAfter(id: str):
"""Decorator for functions to be called after a Hook-Function is called
As with the HookBefore, the hooked up function must accept the same
arguments as the function hooked onto, but also receives a
`hook_result` kwarg containing the result of the function.
"""
def __init__(self, function):
self.function = function
if not id or not isinstance(id, str):
raise TypeError("HookAfter requires the ID of the function to hook up")
def __call__(self, *args, **kwargs):
# Hooks before
for function in _hook_dict[0].get(self.function.__name__, []):
function(*args, **kwargs)
# Main function
ret = self.function(*args, **kwargs)
# Hooks after
for function in _hook_dict[1].get(self.function.__name__, []):
function(*args, **kwargs)
return ret
class HookBefore(object):
"""Decorator for functions to be called before a Hook-Function is called"""
def __init__(self, name):
self.name = name
def __call__(self, function):
_hook_dict[0].setdefault(self.name, []).append(function)
def wrapped(function):
_hooks_after.setdefault(id, []).append(function)
return function
class HookAfter(object):
"""Decorator for functions to be called after a Hook-Function is called"""
def __init__(self, name):
self.name = name
def __call__(self, function):
_hook_dict[1].setdefault(self.name, []).append(function)
return function
return wrapped

3
pyproject.toml Normal file
View File

@ -0,0 +1,3 @@
[build-system]
requires = ["setuptools", "wheel"]
build-backend = "setuptools.build_meta"

View File

@ -1,226 +0,0 @@
#!/usr/bin/python3
from __future__ import annotations # TODO: Remove if python requirement is >= 3.10
import inspect
import argparse
import sys
import pkg_resources
from os import environ
from flaschengeist.app import create_app, install_all
from flaschengeist.config import config
class PrefixMiddleware(object):
def __init__(self, app, prefix=""):
self.app = app
self.prefix = prefix
def __call__(self, environ, start_response):
if environ["PATH_INFO"].startswith(self.prefix):
environ["PATH_INFO"] = environ["PATH_INFO"][len(self.prefix) :]
environ["SCRIPT_NAME"] = self.prefix
return self.app(environ, start_response)
else:
start_response("404", [("Content-Type", "text/plain")])
return ["This url does not belong to the app.".encode()]
class InterfaceGenerator:
known = []
classes = {}
mapper = {
"str": "string",
"int": "number",
"float": "number",
"date": "Date",
"datetime": "Date",
"NoneType": "null",
"bool": "boolean",
}
def __init__(self, namespace, filename):
self.basename = ""
self.namespace = namespace
self.filename = filename
self.this_type = None
def pytype(self, cls):
a = self._pytype(cls)
print(f"{cls} -> {a}")
return a
def _pytype(self, cls):
import typing
origin = typing.get_origin(cls)
arguments = typing.get_args(cls)
if origin is typing.ForwardRef: # isinstance(cls, typing.ForwardRef):
return "", "this" if cls.__forward_arg__ == self.this_type else cls.__forward_arg__
if origin is typing.Union:
print(f"A1: {arguments[1]}")
if len(arguments) == 2 and arguments[1] is type(None):
return "?", self.pytype(arguments[0])[1]
else:
return "", "|".join([self.pytype(pt)[1] for pt in arguments])
if origin is list:
return "", "Array<{}>".format("|".join([self.pytype(a_type)[1] for a_type in arguments]))
name = cls.__name__ if hasattr(cls, "__name__") else cls if isinstance(cls, str) else None
if name is not None:
if name in self.mapper:
return "", self.mapper[name]
else:
return "", name
print(
"WARNING: This python version might not detect all types (try >= 3.9). Could not identify >{}<".format(cls)
)
return "?", "any"
def walker(self, module):
if sys.version_info < (3, 9):
raise RuntimeError("Python >= 3.9 is required to export API")
import typing
if (
inspect.ismodule(module[1])
and module[1].__name__.startswith(self.basename)
and module[1].__name__ not in self.known
):
self.known.append(module[1].__name__)
for cls in inspect.getmembers(module[1], lambda x: inspect.isclass(x) or inspect.ismodule(x)):
self.walker(cls)
elif (
inspect.isclass(module[1])
and module[1].__module__.startswith(self.basename)
and module[0] not in self.classes
and not module[0].startswith("_")
and hasattr(module[1], "__annotations__")
):
self.this_type = module[0]
print("\n\n" + module[0] + "\n")
d = {}
for param, ptype in typing.get_type_hints(module[1], globalns=None, localns=None).items():
if not param.startswith("_") and not param.endswith("_"):
print(f"{param} ::: {ptype}")
d[param] = self.pytype(ptype)
if len(d) == 1:
key, value = d.popitem()
self.classes[module[0]] = value[1]
else:
self.classes[module[0]] = d
def run(self, models):
self.basename = models.__name__
self.walker(("models", models))
def write(self):
with open(self.filename, "w") as file:
file.write("declare namespace {} {{\n".format(self.namespace))
for cls, params in self.classes.items():
if isinstance(params, str):
file.write("\ttype {} = {};\n".format(cls, params))
else:
file.write("\tinterface {} {{\n".format(cls))
for name in params:
file.write("\t\t{}{}: {};\n".format(name, *params[name]))
file.write("\t}\n")
file.write("}\n")
def install(arguments):
app = create_app()
with app.app_context():
install_all()
def run(arguments):
app = create_app()
with app.app_context():
app.wsgi_app = PrefixMiddleware(app.wsgi_app, prefix=config["FLASCHENGEIST"].get("root", ""))
if arguments.debug:
environ["FLASK_DEBUG"] = "1"
app.run(arguments.host, arguments.port, debug=True)
else:
app.run(arguments.host, arguments.port, debug=False)
def export(arguments):
import flaschengeist.models as models
app = create_app()
with app.app_context():
gen = InterfaceGenerator(arguments.namespace, arguments.file)
if not arguments.no_core:
gen.run(models)
if arguments.plugins is not None:
for entry_point in pkg_resources.iter_entry_points("flaschengeist.plugin"):
if len(arguments.plugins) == 0 or entry_point.name in arguments.plugins:
plg = entry_point.load()
if hasattr(plg, "models") and plg.models is not None:
gen.run(plg.models)
gen.write()
def ldap(arguments):
app = create_app()
with app.app_context():
if arguments.set_admin:
from flaschengeist.controller import roleController
from flaschengeist.database import db
role = roleController.get(arguments.set_admin)
role.permissions = roleController.get_permissions()
db.session.commit()
if arguments.sync:
from flaschengeist.controller import userController
from flaschengeist.plugins.auth_ldap import AuthLDAP
from ldap3 import SUBTREE
auth_ldap: AuthLDAP = app.config.get("FG_AUTH_BACKEND")
if auth_ldap is None or not isinstance(auth_ldap, AuthLDAP):
raise Exception("Plugin >auth_ldap< not found")
conn = auth_ldap.ldap.connection
if not conn:
conn = auth_ldap.ldap.connect(auth_ldap.root_dn, auth_ldap.root_secret)
conn.search(auth_ldap.search_dn, "(uid=*)", SUBTREE, attributes=["uid", "givenName", "sn", "mail"])
ldap_users_response = conn.response
for ldap_user in ldap_users_response:
uid = ldap_user["attributes"]["uid"][0]
userController.find_user(uid)
if __name__ == "__main__":
# create the top-level parser
parser = argparse.ArgumentParser()
subparsers = parser.add_subparsers(help="sub-command help", dest="sub_command")
subparsers.required = True
parser_run = subparsers.add_parser("run", help="run flaschengeist")
parser_run.set_defaults(func=run)
parser_run.add_argument("--host", help="set hostname to listen on", default="127.0.0.1")
parser_run.add_argument("--port", help="set port to listen on", type=int, default=5000)
parser_run.add_argument("--debug", help="run in debug mode", action="store_true")
parser_install = subparsers.add_parser(
"install", help="run database setup for flaschengeist and all installed plugins"
)
parser_install.set_defaults(func=install)
parser_export = subparsers.add_parser("export", help="export models to typescript interfaces")
parser_export.set_defaults(func=export)
parser_export.add_argument("--file", help="Filename where to save", default="flaschengeist.d.ts")
parser_export.add_argument("--namespace", help="Namespace of declarations", default="FG")
parser_export.add_argument(
"--no-core",
help="Do not export core declarations (only useful in conjunction with --plugins)",
action="store_true",
)
parser_export.add_argument("--plugins", help="Also export plugins (none means all)", nargs="*")
parser_ldap = subparsers.add_parser("ldap", help="LDAP helper utils")
parser_ldap.set_defaults(func=ldap)
parser_ldap.add_argument('--sync', action="store_true", help="Sync ldap-users with database")
parser_ldap.add_argument('--set-admin', type=str, help="Assign all permissions this to group")
args = parser.parse_args()
args.func(args)

View File

@ -17,6 +17,48 @@ classifiers =
License :: OSI Approved :: MIT License
Operating System :: OS Independent
[options]
include_package_data = True
python_requires = >=3.7
packages = find:
install_requires =
Flask >= 2.0
Pillow>=8.4.0
flask_cors
flask_sqlalchemy>=2.5
sqlalchemy>=1.4.26
toml
werkzeug
PyMySQL;platform_system=='Windows'
mysqlclient;platform_system!='Windows'
[options.extras_require]
argon = argon2-cffi
ldap = flask_ldapconn; ldap3
test = pytest; coverage
[options.package_data]
* = *.toml
[options.entry_points]
console_scripts =
flaschengeist = flaschengeist.cli:cli
flask.commands =
ldap = flaschengeist.plugins.auth_ldap.cli:ldap
users = flaschengeist.plugins.users.cli:users
flaschengeist.plugins =
# Authentication providers
auth_plain = flaschengeist.plugins.auth_plain:AuthPlain
auth_ldap = flaschengeist.plugins.auth_ldap:AuthLDAP [ldap]
# Route providers (and misc)
auth = flaschengeist.plugins.auth:AuthRoutePlugin
users = flaschengeist.plugins.users:UsersPlugin
roles = flaschengeist.plugins.roles:RolesPlugin
balance = flaschengeist.plugins.balance:BalancePlugin
mail = flaschengeist.plugins.message_mail:MailMessagePlugin
pricelist = flaschengeist.plugins.pricelist:PriceListPlugin
scheduler = flaschengeist.plugins.scheduler:SchedulerPlugin
[bdist_wheel]
universal = True

View File

@ -1,77 +0,0 @@
from setuptools import setup, find_packages, Command
import subprocess
import os
mysql_driver = "PyMySQL" if os.name == "nt" else "mysqlclient"
class DocsCommand(Command):
description = "Generate and export API documentation using pdoc3"
user_options = [
# The format is (long option, short option, description).
("output=", "o", "Documentation output path"),
]
def initialize_options(self):
self.output = "./docs"
def finalize_options(self):
pass
def run(self):
"""Run command."""
command = [
"python",
"-m",
"pdoc",
"--skip-errors",
"--html",
"--output-dir",
self.output,
"flaschengeist",
]
self.announce(
"Running command: %s" % str(command),
)
subprocess.check_call(command)
setup(
packages=find_packages(),
package_data={"": ["*.toml"]},
scripts=["run_flaschengeist"],
python_requires=">=3.7",
install_requires=[
"Flask >= 2.0",
"toml",
"sqlalchemy>=1.4.26",
"flask_sqlalchemy>=2.5",
"flask_cors",
"Pillow>=8.4.0",
"werkzeug",
mysql_driver,
],
extras_require={
"ldap": ["flask_ldapconn", "ldap3"],
"argon": ["argon2-cffi"],
"test": ["pytest", "coverage"],
},
entry_points={
"flaschengeist.plugins": [
# Authentication providers
"auth_plain = flaschengeist.plugins.auth_plain:AuthPlain",
"auth_ldap = flaschengeist.plugins.auth_ldap:AuthLDAP [ldap]",
# Route providers (and misc)
"auth = flaschengeist.plugins.auth:AuthRoutePlugin",
"users = flaschengeist.plugins.users:UsersPlugin",
"roles = flaschengeist.plugins.roles:RolesPlugin",
"balance = flaschengeist.plugins.balance:BalancePlugin",
"mail = flaschengeist.plugins.message_mail:MailMessagePlugin",
"pricelist = flaschengeist.plugins.pricelist:PriceListPlugin",
"scheduler = flaschengeist.plugins.scheduler:SchedulerPlugin",
],
},
cmdclass={
"docs": DocsCommand,
},
)