Milliways/lenticular_cloud2
4
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity

add more registration stuff, admin interface

Browse source
This commit is contained in:
TuxCoder 2020-06-01 23:43:10 +02:00
parent c1c8876b63
commit 67b69104d6
23 changed files with 369 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

3
lenticular_cloud/app.py
View file

@ -54,12 +54,13 @@ def init_app(name=None):
hydra_client = hydra.ApiClient(hydra_config)
app.hydra_api = hydra.AdminApi(hydra_client)
from .views import auth_views, frontend_views, init_login_manager, api_views, pki_views
from .views import auth_views, frontend_views, init_login_manager, api_views, pki_views, admin_views
init_login_manager(app)
app.register_blueprint(auth_views)
app.register_blueprint(frontend_views)
app.register_blueprint(api_views)
app.register_blueprint(pki_views)
app.register_blueprint(admin_views)
@app.before_request
def befor_request():

16
lenticular_cloud/form/auth.py
View file

@ -6,7 +6,7 @@ from wtforms import StringField, SubmitField, TextField, \
SelectField, Form as NoCsrfForm, SelectMultipleField
from wtforms.fields.html5 import EmailField
from wtforms.widgets.html5 import NumberInput, DateInput
from wtforms.validators import DataRequired, NumberRange, Optional, NoneOf, Length
from wtforms.validators import DataRequired, NumberRange, Optional, NoneOf, Length, Regexp
from datetime import datetime
@ -33,12 +33,18 @@ class Fido2Form(FlaskForm):
class ConsentForm(FlaskForm):
# scopes = SelectMultipleField(gettext('scopes'))
# audiences = SelectMultipleField(gettext('audiences'))
remember = BooleanField(gettext('remember me'))
remember = BooleanField(gettext('remember'))
submit = SubmitField()
class RegistrationForm(FlaskForm):
username = StringField(gettext('Username'), validators=[DataRequired()])
password = PasswordField(gettext('Password'), validators=[DataRequired()])
alternative_email = EmailField(gettext('Alternative Email'))
username = StringField(gettext('Username'), validators=[
DataRequired(),
Regexp('^[a-zA-Z0-9-.]+$', message=gettext('Only `a-z`, `A-Z`, `0-9`, `-.` is allowed for username'))
])
password = PasswordField(gettext('Password'), validators=[
DataRequired(),
Length(min=6)
])
alternative_email = EmailField(gettext('Alternative Email'), render_kw={"placeholder": "Optional"})
submit = SubmitField()

87
lenticular_cloud/model.py
View file

@ -28,6 +28,14 @@ base_dn = ''
db = SQLAlchemy() # type: SQLAlchemy
class UserSignUp(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String, nullable=False)
password = db.Column(db.String, nullable=False)
alternative_email = db.Column(db.String)
created_at = db.Column(db.DateTime, nullable=False,
default=datetime.now)
class SecurityUser(UserMixin):
@ -51,7 +59,7 @@ class EntryBase(db.Model):
__abstract__ = True # for sqlalchemy
_type = None # will get replaced by the local type
_query_object = None # will get replaced by the local type
_ldap_query_object = None # will get replaced by the local type
_base_dn = LambdaStr(lambda: base_dn)
# def __init__(self, ldap_object=None, **kwargs):
@ -60,39 +68,45 @@ class EntryBase(db.Model):
# else:
# self._ldap_object = ldap_object
def __str__(self):
def __str__(self) -> str:
return str(self._ldap_object)
@classmethod
def get_object_def(cls):
def get_object_def(cls) -> ObjectDef:
return ObjectDef(cls.object_classes, ldap_conn)
@classmethod
def get_base(cls):
def get_entry_type(cls) -> EntryType:
return EntryType(cls.get_dn(), cls.object_classes, ldap_conn)
@classmethod
def get_base(cls) -> str:
return cls.base_dn.format(_base_dn=base_dn)
@classmethod
def get_dn(cls) -> str:
return cls.dn.replace('{base_dn}', cls.get_base())
@classmethod
def get_type(cls):
if cls._type is None:
cls._type = EntryType(cls.dn.replace('{base_dn}',cls.get_base()), cls.object_classes, ldap_conn)
cls._type = EntryType(cls.get_dn(), cls.object_classes, ldap_conn)
return cls._type
def commit(self):
def ldap_commit(self):
self._ldap_object.entry_commit_changes()
def add(self):
print(self._ldap_object.entry_attributes_as_dict)
def ldap_add(self):
ret = ldap_conn.add(
self.dn, self.object_classes, self._ldap_object.entry_attributes_as_dict)
logger.debug(ret)
pass
self.entry_dn, self.object_classes, self._ldap_object.entry_attributes_as_dict)
if not ret:
raise Exception('ldap error')
@classmethod
def query_(cls):
if cls._query_object is None:
cls._query_object = cls._query(cls)
return cls._query_object
if cls._ldap_query_object is None:
cls._ldap_query_object = cls._query(cls)
return cls._ldap_query_object
class _query(object):
def __init__(self, clazz):
@ -114,8 +128,6 @@ class EntryBase(db.Model):
return self._query(None)
class Service(object):
def __init__(self, name):
@ -213,16 +225,20 @@ class User(EntryBase):
id = db.Column(
db.String(length=36), primary_key=True, default=generate_uuid)
username = db.Column(
db.String, unique=True)
db.String, unique=True, nullable=False)
created_at = db.Column(db.DateTime, nullable=False,
default=datetime.now)
modified_at = db.Column(db.DateTime, nullable=False,
default=datetime.now, onupdate=datetime.now)
last_login = db.Column(db.DateTime, nullable=True)
totps = db.relationship('Totp', back_populates='user')
dn = "uid={uid},{base_dn}"
base_dn = "ou=users,{_base_dn}"
object_classes = ["top", "inetOrgPerson", "LenticularUser"]
def __init__(self,**kwargs):
def __init__(self, **kwargs):
self._ldap_object = None
super(db.Model).__init__(**kwargs)
@ -236,6 +252,13 @@ class User(EntryBase):
def make_writeable(self):
self._ldap_object = self._ldap_object.entry_writable()
@property
def groups(self):
if self.username == 'tuxcoder':
return [Group(name='admin')]
else:
return []
@property
def entry_dn(self):
return self._ldap_object.entry_dn
@ -274,7 +297,7 @@ class User(EntryBase):
self.make_writeable()
password_hashed = crypt.crypt(password_new)
self._ldap_object.userPassword = ('{CRYPT}' + password_hashed).encode()
self.commit()
self.ldap_commit()
class _query(EntryBase._query):
@ -296,6 +319,21 @@ class User(EntryBase):
else:
return None
@staticmethod
def new(user_data: UserSignUp):
user = User()
user.username = user_data.username.lower()
domain = current_app.config['DOMAIN']
ldap_object = User.get_entry_type()(
uid=user_data.username.lower(),
sn=user_data.username,
cn=user_data.username,
userPassword='{CRYPT}' + user_data.password,
mail=f'{user_data.username}@{domain}')
user._ldap_object = ldap_object
user.ldap_add()
return user
class Totp(db.Model):
@ -321,9 +359,6 @@ class Group(EntryBase):
fullname = AttrDef("cn")
class UserSignUp(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String, nullable=False)
password = db.Column(db.String, nullable=False)
alternative_email = db.Column(db.String)
name = db.Column(db.String(), nullable=False, unique=True)

1
lenticular_cloud/views/__init__.py
View file

@ -2,5 +2,6 @@
from .auth import auth_views
from .frontend import frontend_views, init_login_manager
from .admin import admin_views
from .api import api_views
from .pki import pki_views

66
lenticular_cloud/views/admin.py Normal file
View file

@ -0,0 +1,66 @@
import flask
from flask import Blueprint, redirect, request, url_for, render_template
from flask import current_app, session
from flask import jsonify
from flask_login import current_user, logout_user
from oauthlib.oauth2.rfc6749.errors import TokenExpiredError
from ..model import db, User, UserSignUp
admin_views = Blueprint('admin', __name__, url_prefix='/admin')
def before_request():
try:
resp = current_app.oauth.session.get('/userinfo')
data = resp.json()
if not current_user.is_authenticated or resp.status_code is not 200:
logout_user()
return redirect(url_for('oauth.login'))
if 'admin' not in data['groups']:
return 'Not an admin', 403
except TokenExpiredError:
logout_user()
return redirect(url_for('oauth.login'))
admin_views.before_request(before_request)
@admin_views.route('/', methods=['GET', 'POST'])
def index():
return render_template('admin/index.html.j2')
@admin_views.route('/user', methods=['GET'])
def users():
users = User.query.all()
return render_template('admin/users.html.j2', users=users)
@admin_views.route('/registrations', methods=['GET'])
def registrations():
users = UserSignUp.query.all()
return render_template('admin/registrations.html.j2', users=users)
@admin_views.route('/registration/<registration_id>', methods=['DELETE'])
def registration_delete(registration_id):
user_data = UserSignUp.query.get(registration_id)
if user_data is None:
return jsonify({}), 404
db.session.delete(user_data)
db.session.commit()
return jsonify({})
@admin_views.route('/registration/<registration_id>', methods=['PUT'])
def registration_accept(registration_id):
user_data = UserSignUp.query.get(registration_id)
#create user
user = User.new(user_data)
db.session.add(user)
db.session.delete(user_data)
db.session.commit()
return jsonify({})

41
lenticular_cloud/views/auth.py
View file

@ -7,13 +7,15 @@ from flask import current_app, session
from flask.templating import render_template
from flask_babel import gettext
from flask import request, url_for
from flask import request, url_for, jsonify
from flask_login import login_required, login_user, logout_user, current_user
import logging
from urllib.parse import urlparse
from base64 import b64decode, b64encode
import http
import crypt
import ory_hydra_client
from datetime import datetime
from ..model import db, User, SecurityUser, UserSignUp
from ..form.auth import ConsentForm, LoginForm, RegistrationForm
@ -31,8 +33,11 @@ def consent():
form = ConsentForm()
remember_for = 60*60*24*30 # remember for 7 days
consent_request = current_app.hydra_api.get_consent_request(
request.args['consent_challenge'])
try:
consent_request = current_app.hydra_api.get_consent_request(
request.args['consent_challenge'])
except ory_hydra_client.exceptions.ApiException:
return redirect(url_for('frontend.index'))
requested_scope = consent_request.requested_scope
requested_audiences = consent_request.requested_access_token_audience
@ -40,9 +45,11 @@ def consent():
if form.validate_on_submit() or consent_request.skip:
user = User.query.get(consent_request.subject)
token_data = {
'name': str(user.username),
'preferred_username': str(user.username),
'email': str(user.email),
'email_verified': True,
'groups': [group.name for group in user.groups]
}
id_token_data = {}
if 'openid' in requested_scope:
@ -70,7 +77,10 @@ def consent():
@auth_views.route('/login', methods=['GET', 'POST'])
def login():
login_challenge = request.args.get('login_challenge')
login_request = current_app.hydra_api.get_login_request(login_challenge)
try:
login_request = current_app.hydra_api.get_login_request(login_challenge)
except ory_hydra_client.exceptions.ApiValueError:
return redirect(url_for('frontend.index'))
if login_request.skip:
resp = current_app.hydra_api.accept_login_request(
@ -93,7 +103,11 @@ def login():
@auth_views.route('/login/auth', methods=['GET', 'POST'])
def login_auth():
login_challenge = request.args.get('login_challenge')
login_request = current_app.hydra_api.get_login_request(login_challenge)
try:
login_request = current_app.hydra_api.get_login_request(login_challenge)
except ory_hydra_client.exceptions.ApiValueError:
return redirect(url_for('frontend.index'))
if 'username' not in session:
return redirect(url_for('auth.login'))
auth_forms = {}
@ -115,7 +129,8 @@ def login_auth():
# db.session.commit()
subject = user.id
user.last_login = datetime.now()
db.session.commit()
resp = current_app.hydra_api.accept_login_request(
login_challenge, body={
'subject': subject,
@ -135,8 +150,13 @@ def logout():
@auth_views.route("/sign_up", methods=["GET", "POST"])
@auth_views.route("/sign_up", methods=["GET"])
def sign_up():
form = RegistrationForm()
return render_template('auth/sign_up.html.j2', form=form)
@auth_views.route("/sign_up", methods=["POST"])
def sign_up_submit():
form = RegistrationForm()
if form.validate_on_submit():
user = UserSignUp()
@ -145,5 +165,8 @@ def sign_up():
user.alternative_email = form.data['alternative_email']
db.session.add(user)
db.session.commit()
return render_template('auth/sign_up.html.j2', form=form)
return jsonify({})
return jsonify({
'status': 'error',
'errors': form.errors
})

2
lenticular_cloud/views/frontend.py
View file

@ -27,8 +27,10 @@ def before_request():
try:
resp = current_app.oauth.session.get('/userinfo')
if not current_user.is_authenticated or resp.status_code is not 200:
logout_user()
return redirect(url_for('oauth.login'))
except TokenExpiredError:
logout_user()
return redirect(url_for('oauth.login'))