<commit_before># -*- coding:Utf-8 -*-


import os, sys, stat, re
import config

from io import open
from ldap3 import Server, Connection, ALL
from flask import request
from sqlalchemy import exc
from sqlalchemy.orm import sessionmaker
from app import app, db
from app.models_mod import user, target
from . import api
from .. import utilities as utils


def useruid(s, login):
    """Connect to a LDAP and check the uid matching the given field data"""
    uid = False
    c = Connection(s, config.LDAPACC, 
                   password=config.LD<PASSWORD>, auto_bind=True)

    if c.result["description"] != "success":
        app.logger.error("Error connecting to the LDAP with the service account")
        return False

    # Look for the user entry.
    if not c.search(config.LDAPBASE,
                    "(" + config.LDAPFIELD + "=" + login + ")") :
        app.logger.error("Error: Connection to the LDAP with service account failed")
    else:
        if len(c.entries) >= 1 :
            if len(c.entries) > 1 :
                app.logger.error("Error: multiple entries with this login. "+ \
                          "Trying first entry...")
            uid = c.entries[0].entry_dn
        else:
            app.logger.error("Error: Login not found")
        c.unbind()
    
    return uid


def try_ldap_login(login, password):
    """ Connect to a LDAP directory to verify user login/passwords"""
    result = "Wrong login/password"
    s = Server(config.LDAPURI, port=config.LDAPPORT,
               use_ssl=False, get_info=ALL)
    # 1. connection with service account to find the user uid
    uid = useruid(s, login)
   
    if uid: 
        # 2. Try to bind the user to the LDAP
        c = Connection(s, user = uid , password = password, auto_bind = True)
        c.open()
        c.bind()
        result =  c.result["description"] # "success" if bind is ok
        c.unbind()

    return result


def try_login(login, password, method="LDAP"):
    if method == "LDAP":
        return try_ldap_login(login, password)


@app.route("/user/login", methods=["POST"])
def user_login():
    """Allow passhportd to handle login/passwords for users"""
    # Only POST data are handled
    if request.method != "POST":
        return utils.response("ERROR: POST method is required ", 405)

    # Simplification for the reading
    login = request.form["login"]
    password = request.form["password"]

    # Check for required fields
    if not login or not password:
        return utils.response("ERROR: The login and password are required ", 417)

    # Check data validity uppon LDAP/local/whatever...
    result = try_login(login, password)
    if result == "success":
        app.logger.info("Authentication ok for {}".format(login))
        # If the LDAP connection is ok, user can connect
        return utils.response("Authorized", 200)
    app.logger.warning("Authentication error for {} => ".format(login) + str(result))
    return utils.response("Refused: " + str(result), 200)


@app.route("/user/list")
def user_list():<commit_msg>Update user.py<commit_after># -*- coding:Utf-8 -*-


import os, sys, stat, re
import config

from io import open
from ldap3 import Server, Connection, ALL
from ldap3.utils.dn import escape_rdn
from flask import request
from sqlalchemy import exc
from sqlalchemy.orm import sessionmaker
from app import app, db
from app.models_mod import user, target
from . import api
from .. import utilities as utils


def useruid(s, login):
    """Connect to a LDAP and check the uid matching the given field data"""
    uid = False
    c = Connection(s, config.LDAPACC, 
                   password=config.LDAPPASS, auto_bind=True)

    if c.result["description"] != "success":
        app.logger.error("Error connecting to the LDAP with the service account")
        return False

    # Look for the user entry.
    if not c.search(config.LDAPBASE,
                    "(" + config.LDAPFIELD + "=" + login + ")") :
        app.logger.error("Error: Connection to the LDAP with service account failed")
    else:
        if len(c.entries) >= 1 :
            if len(c.entries) > 1 :
                app.logger.error("Error: multiple entries with this login. "+ \
                          "Trying first entry...")
            uid = c.entries[0].entry_dn
        else:
            app.logger.error("Error: Login not found")
        c.unbind()
    
    return uid


def try_ldap_login(login, password):
    """ Connect to a LDAP directory to verify user login/passwords"""
    result = "Wrong login/password"
    s = Server(config.LDAPURI, port=config.LDAPPORT,
               use_ssl=False, get_info=ALL)
    # 1. connection with service account to find the user uid
    uid = useruid(s, login)
   
    if uid: 
        # 2. Try to bind the user to the LDAP
        c = Connection(s, user = uid , password = password, auto_bind = True)
        c.open()
        c.bind()
        result =  c.result["description"] # "success" if bind is ok
        c.unbind()

    return result


def try_login(login, password, method="LDAP"):
    if method == "LDAP":
        return try_ldap_login(login, password)


@app.route("/user/login", methods=["POST"])
def user_login():
    """Allow passhportd to handle login/passwords for users"""
    # Only POST data are handled
    if request.method != "POST":
        return utils.response("ERROR: POST method is required ", 405)

    # Simplification for the reading
    login = request.form["login"]
    password = request.form["password"]

    # Check for required fields
    if not login or not password:
        return utils.response("ERROR: The login and password are required ", 417)
    elif login != escape_rdn(login):
        return utils.response("ERROR: Bad input", 417)

    # Check data validity uppon LDAP/local/whatever...
    result = try_login(login, password)
    if result == "success":
        app.logger.info("Authentication ok for {}".format(login))
        # If the LDAP connection is ok, user can connect
        return utils.response("Authorized", 200)
    app.logger.warning("Authentication error for {} => ".format(login) + str(result))
    return utils.response("Refused: " + str(result), 200)


@app.route("/user/list")
def user_list():
{'fraction_non_alphanumeric': 0.11246590726776833, 'fraction_numerical': 0.006738328252847746, 'mean_word_length': 2.6307513104251603, 'pattern_counts': {'":': 8, '<': 4, '<?xml version=': 0, '>': 10, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2961786', 'n_tokens_mistral': 1844, 'n_tokens_neox': 1771, 'n_words': 606}