import ldap from ldap3 import Server, Connection from flask_wtf import FlaskForm from flask_login import UserMixin from ldap3.core.exceptions import LDAPBindError from wtforms import StringField, PasswordField, BooleanField, SubmitField from wtforms.validators import DataRequired from accounts import app, db def get_ldap_connection(): server = Server(app.config['LDAP_HOST']) conn = Connection(server, app.config['LDAP_USERNAME'], app.config['LDAP_PASSWORD'], auto_bind=True) return conn class User(db.Model): __tablename__ = 'user' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(100)) password = db.Column(db.String(128)) authenticated = db.Column(db.Boolean, default=False) def __init__(self, username, password): self.username = username self.password = password @staticmethod def try_login(username, password): conn = get_ldap_connection() conn.search(app.config['LDAP_BASE_DN'], app.config['LDAP_USER_OBJECT_FILTER'] % username, attributes=['*']) if len(conn.entries) > 0: Connection(app.config['LDAP_HOST'], conn.entries[0].entry_dn, password, auto_bind=True) return raise LDAPBindError def is_authenticated(self): return self.authenticated def is_active(self): return True def is_anonymous(self): return False def get_id(self): return self.id def get_user_dict(self): user = {'dn': '', 'firstName': '', 'lastName': '', 'email': '', 'userName': self.username, } conn = get_ldap_connection() conn.search(app.config['LDAP_BASE_DN'], app.config['LDAP_USER_OBJECT_FILTER'] % self.username, attributes=['*']) user['dn'] = conn.entries[0].entry_dn user['firstName'] = conn.entries[0].givenName.value user['lastName'] = conn.entries[0].sn.value user['email'] = conn.entries[0].mail.value return user class LoginForm(FlaskForm): username = StringField('Username', validators=[DataRequired()]) password = PasswordField('Password', validators=[DataRequired()]) remember_me = BooleanField('Remember Me') submit = SubmitField('Sign In')