crans/lc_ldap
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
lc_ldap/lc_ldap.py
Pierre-Elliott Bécue 42c48f77e8 Quelques améliorations, mise en place de fonctions pour tester parenté, 
tester si c'est soi-même qu'on modifie, pour tester si on peut bien
altérer l'objet concerné.
Création des objets en deux temps (on crée l'objet Crans, puis
on l'enregistre dans ldap après test des droits.
Changement de méthode de binding : le binding nominatif va être bien
trop complexe à implémenter, on va donc faire autrement...
2013-01-28 00:45:01 +01:00

889 lines
35 KiB
Python
Raw Blame History

#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# LC_LDAP.PY-- LightWeight CransLdap
#
# Copyright (C) 2010-2013 Cr@ns <roots@crans.org>
# Authors: Antoine Durand-Gasselin <adg@crans.org>
# Nicolas Dandrimont <olasd@crans.org>
# Olivier Iffrig <iffrig@crans.org>
# Valentin Samir <samir@crans.org>
# Daniel Stan <dstan@crans.org>
# Vincent Le Gallic <legallic@crans.org>
# Pierre-Elliott Bécue <becue@crans.org>
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# * Neither the name of the Cr@ns nor the names of its contributors may
# be used to endorse or promote products derived from this software
# without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT
# HOLDER> BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
from __future__ import with_statement
import os
import sys
import ldap
import ldap.filter
from ldap.modlist import addModlist, modifyModlist
import re
import netaddr
import datetime
import copy
import time
import random
try:
from Levenshtein import jaro
except ImportError:
def jaro(a, b): return 0
sys.path.append('/usr/scripts/gestion')
import config
import crans_utils
from attributs import *
from ldap_locks import CransLock
import ridtools
uri = 'ldap://ldap.adm.crans.org/'
base_dn = 'ou=data,dc=crans,dc=org'
log_dn = "cn=log"
# Protection contre les typos
created = 'created'
modified = 'modified'
deleted = 'deleted'
# Pour enregistrer dans l'historique, on a besoin de savoir qui exécute le script
# Si le script a été exécuté via un sudo, la variable SUDO_USER (l'utilisateur qui a effectué le sudo)
# est plus pertinente que USER (qui sera root)
current_user = os.getenv("SUDO_USER") or os.getenv("USER")
# Quand on a besoin du fichier de secrets
def import_secrets():
if not "/etc/crans/secrets/" in sys.path:
sys.path.append("/etc/crans/secrets/")
import secrets
return secrets
# Champs à ignorer dans l'historique
HIST_IGNORE_FIELDS = ["modifiersName", "entryCSN", "modifyTimestamp", "historique"]
def escape(chaine):
"""Renvoie une chaîne échapée pour pouvoir la mettre en toute sécurité
dans une requête ldap."""
return ldap.filter.escape_filter_chars(chaine)
def ldif_to_uldif(ldif):
"""
Transforme un dictionnaire ldif en un dictionnaire
ldif unicode.
"""
uldif = {}
for attr, vals in ldif.items():
uldif[attr] = [ unicode(val, 'utf-8') for val in vals ]
return uldif
def ldif_to_cldif(ldif, conn, check_ctxt = True):
"""Transforme un dictionnaire renvoyé par python-ldap, en
un dictionnaire dont les valeurs sont des instances de Attr
Lorsqu'on récupère le ldif de la base ldap, on n'a pas besoin
de faire de tests...
"""
cldif = {}
for attr, vals in ldif.items():
cldif[attr] = [ attrify(val, attr, ldif, conn, check_ctxt) for val in vals]
return cldif
def cldif_to_ldif(cldif):
"""
Transforme un ldif crans en ldif valide au sens openldap.
Ce ldif est celui qui sera transmis à la base.
"""
ldif = {}
for attr, vals in cldif.items():
ldif[attr] = [ str(val) for val in vals ]
return ldif
def lc_ldap_test():
"""Binding LDAP à la base de tests"""
return lc_ldap(uri='ldap://vo.adm.crans.org',dn='cn=admin,dc=crans,dc=org', cred='75bdb64f32')
def lc_ldap_admin():
"""Binding LDAP à la vraie base, en admin.
Possible seulement si on peut lire secrets.py"""
secrets = import_secrets()
return lc_ldap(uri='ldap://ldap.adm.crans.org/', dn=secrets.ldap_auth_dn, cred=secrets.ldap_password)
class lc_ldap(ldap.ldapobject.LDAPObject):
"""Connexion à la base ldap crans, chaque instance représente une connexion
"""
def __init__(self, dn=None, user=None, cred=None, uri=uri, test=False):
"""Initialise la connexion ldap,
- En authentifiant avec dn et cred s'ils sont précisés
- Si dn n'est pas précisé, mais que user est précisé, récupère
le dn associé à l'uid user, et effectue l'authentification
avec ce dn et cred
- Sinon effectue une authentification anonyme
Si test est à True, on se connecte à la base de test sur vo.
"""
if test:
uri = "ldapi:///"
ldap.ldapobject.LDAPObject.__init__(self, uri)
if user and not re.match('[a-z_][a-z0-9_-]*', user):
raise ValueError('Invalid user name: %r' % user)
# Si un username, on récupère le dn associé…
if user and not dn:
if test:
# …en anonyme si on se connecte à la base de test
self.simple_bind_s(base_dn)
else:
# …sinon, en se connectant en readonly (on récupère les identifiants dans secrets.py)
secrets = import_secrets()
self.simple_bind_s(who=secrets.ldap_readonly_auth_dn, cred=secrets.ldap_readonly_password)
res = self.search_s(base_dn, 1, 'uid=%s' % user)
if len(res) < 1:
raise ldap.INVALID_CREDENTIALS({'desc': 'No such user: %s' % user })
elif len(res) > 1:
raise ldap.INVALID_CREDENTIALS({'desc': 'Too many matches: uid=%s' % user })
else:
dn = res[0][0]
# Si on a un dn, on se connecte avec à la base ldap sinon on s'y
# connecte en anonyme
if dn:
self.conn = self.bind_s(secrets.ldap_auth_dn, secrets.ldap_password)
self.dn = dn
self.droits = self.search_s(dn, ldap.SCOPE_BASE, attrlist=['droits'])[0][1].get('droits', [])
else:
self.conn = self.simple_bind_s()
self.dn = None
self.droits = []
def search(self, filterstr='(objectClass=*)', mode='ro', dn= base_dn, scope=ldap.SCOPE_SUBTREE, sizelimit=1000):
"""La fonction de recherche dans la base ldap qui renvoie un liste de
CransLdapObjects. On utilise la feature de sizelimit de python ldap"""
ldap_res = self.search_ext_s(dn, scope, filterstr, sizelimit=sizelimit)
ret = []
for dn, ldif in ldap_res:
ret.append(new_cransldapobject(self, dn, mode, ldif))
return ret
def allMachinesAdherents(self):
"""Renvoie la liste de toutes les machines et de tous les adherents
(club et Association Crans compris). Conçue pour s'éxécuter le plus
rapidement possible. On dumpe malgré tout toute la base."""
res = {}
parent = {}
machines = {}
# (proxying de la base ldap)
for dn, attrs in self.search_s(base_dn, scope=2):
# On crée les listes des machines et propriétaires
if dn.startswith('mid='): # les machines
m = new_cransldapobject(self, dn, ldif = attrs)
parent_dn = dn.split(',', 1)[1]
if not machines.has_key(parent_dn):
machines[parent_dn]=[]
machines[parent_dn].append(m)
elif (dn.startswith('aid=') or dn.startswith('cid=') or dn==base_dn) and not parent.has_key(dn):
parent[dn]=new_cransldapobject(self, dn, ldif = attrs)
allmachines=[]
for dn,mlist in machines.items(): # on associe propriétaires et machines
parent[dn]._machines=mlist
for m in mlist:
m._proprio=parent[dn]
allmachines.append(m)
return allmachines,parent.values() # on renvoie la liste des machines et des adherents (dont club et crans)
def allMachines(self):
"""Renvoie la liste de toutes les machines, Conçue pour
s'éxécuter le plus rapidement possible. On dumpe malgré tout
toute la base, c'est pour pouvoir aussi rajouter à moindre coût
les propriétaires."""
machines,_=self.allMachinesAdherents()
return machines
def allAdherents(self):
"""Renvoie la liste de toutes les adherents, Conçue pour
s'éxécuter le plus rapidement possible. On dumpe malgré tout
toute la base, c'est pour pouvoir aussi rajouter à moindre coût
les machines."""
_,adherents=self.allMachinesAdherents()
return adherents
def newMachine(self, parent, realm, uldif):
"""Crée une nouvelle machine: realm peut être:
fil, fil-v6, wifi, wifi-v6, adm, gratuit, personnel-ens, special
--Partiellement implémenté"""
#adm, serveurs, bornes, wifi, adherents, gratuit ou personnel-ens"""
owner = self.search('objectClass=*', dn=parent, scope=0)[0]
if realm in ["adm", "serveurs", "serveurs-v6", "adm-v6"]:
uldif['objectClass'] = [u'machineCrans']
assert isinstance(owner, AssociationCrans)
elif realm == "bornes":
uldif['objectClass'] = [u'borneWifi']
assert isinstance(owner, AssociationCrans)
elif realm in ["wifi", "wifi-v6"]:
uldif['objectClass'] = [u'machineWifi']
assert isinstance(owner, adherent) or isinstance(owner, club)
elif realm in ["fil", "fil-v6", "personnel-ens"]:
uldif['objectClass'] = [u'machineFixe']
assert isinstance(owner, adherent) or isinstance(owner, club)
else: raise ValueError("Realm inconnu: %r" % realm)
# On récupère le premier id libre dans la plages s'il n'est pas
# déjà précisé dans le ldiff
rid = uldif.setdefault('rid', [ unicode(self._find_id('rid', plage)) ])
# La machine peut-elle avoir une ipv4 ?
if 'v6' not in realm:
uldif['ipHostNumber'] = [ unicode(crans_utils.ip4_of_rid(rid[0])) ]
uldif['ip6HostNumber'] = [ unicode(crans_utils.ip6_of_mac(uldif['macAddress'][0], rid[0])) ]
# On récupère la plage des mids
plage = xrange( *(config.rid[realm]))
# Tout doit disparaître !!
machine = self._create_entity('mid=%s,%s' % (mid[0], parent), uldif)
if machine.may_be(created, self.droits + self._is_parent(machine)):
machine.create()
else:
raise EnvironmentError("Vous n'avez pas le droit de créer cette machine.")
def newAdherent(self, uldif):
"""Crée un nouvel adhérent"""
aid = uldif.setdefault('aid', [ unicode(self._find_id('aid')) ])
uldif['objectClass'] = [u'adherent']
adherent = self._create_entity('aid=%s,%s' % (aid[0], base_dn), uldif)
if adherent.may_be(created, self):
adherent.create()
else:
raise EnvironmentError("Vous n'avez pas le droit de créer cet adhérent.")
def newClub(self, uldif):
"""Crée un nouveau club"""
cid = uldif.setdefault('cid', [ unicode(self._find_id('cid')) ])
uldif['objectClass'] = [u'club']
club = self._create_entity('cid=%s,%s' % (cid[0], base_dn), uldif)
if club.may_be(created, self):
club.create()
else:
raise EnvironmentError("Vous n'avez pas le droit de créer cet adhérent.")
def newFacture(self, uldif):
"""Crée une nouvelle facture
--Non implémenté !"""
raise NotImplementedError()
def _create_entity(self, dn, uldif):
'''Crée une nouvelle entité ldap en dn, avec attributs ldif:
uniquement en unicode'''
# Conversion en cldiff pour vérification des valeurs
cldif = ldif_to_cldif(uldif, self)
# Conversion en ascii
ldif = cldif_to_ldif(cldif)
# Renvoi du CransLdapObject
return new_cransldapobject(self, dn, 'rw', ldif)
def _find_id(self, attr, plage=None):
'''Trouve un id libre. Si une plage est fournie, cherche
l'id dans celle-ci, sinon, prend le plus élevé possible.'''
res = self.search_s(base_dn, ldap.SCOPE_SUBTREE, '%s=*' % attr, attrlist = [attr])
nonfree = [ int(r[1].get(attr)[0]) for r in res if r[1].get(attr) ]
nonfree.sort()
if plage != None:
for i in plage:
if i in nonfree:
continue
else:
break
else:
raise EnvironmentError('Aucun %s libre dans la plage [%d, %d]' %
(attr, plage[0], i))
else:
i = nonfree[-1]+1
return i
def _is_parent(self, obj):
"""
Teste le lien de parenté de l'objet 1 sur l'objet 2.
Retourne une liste qui s'ajoutera à la liste des droits
"""
if obj.dn.endswith(self.dn) and obj.dn != self.dn:
return [parent]
else:
return []
def _is_self(obj1, obj2):
"""
Teste si l'objet qui se binde est celui qui est en cours de modif.
Retourne une liste qui s'ajoutera à la liste des droits
"""
if obj.dn == self.dn:
return [soi]
else:
return []
def new_cransldapobject(conn, dn, mode='ro', ldif = None):
"""Crée un objet CransLdap en utilisant la classe correspondant à
l'objectClass du ldif
--pour usage interne à la libraire uniquement !"""
classe = None
if dn == base_dn:
classe = AssociationCrans
elif ldif:
classe = globals()[ldif['objectClass'][0]]
else:
res = conn.search_s(dn, 0)
if not res:
raise ValueError ('objet inexistant: %s' % dn)
_, attrs = res[0]
classe = globals()[attrs['objectClass'][0]]
return classe(conn, dn, mode, ldif)
class CransLdapObject(object):
"""Classe de base des objets CransLdap.
Cette classe ne devrait pas être utilisée directement."""
can_be_by = { created: [nounou],
modified: [nounou],
deleted: [nounou],
}
""" Champs uniques et nécessaires """
ufields = []
""" Champs uniques facultatifs """
ofields = []
""" Champs multivalués facultatifs """
mfields = []
""" Champs obligatoires multivalué """
xfields = []
def __init__(self, conn, dn, mode='ro', ldif = None):
'''
Créée une instance d'un objet Crans (machine, adhérent,
etc...) à ce dn, si ldif est précisé, n'effectue pas de
recherche dans la base ldap.
'''
self.mode = mode
self.attrs = {} # Contient un dico uldif qui doit représenter ce qui
# est dans la base
self._modifs = None # C'est là qu'on met les modifications
if not isinstance(conn, lc_ldap):
raise TypeError("conn doit être une instance de lc_ldap")
self.conn = conn
self.dn = dn
orig = {}
if ldif:
self.attrs = ldif
if dn != base_dn: # new_cransldapobject ne donne pas de ldif formaté et utilise un ldif non formaté, donc on formate
self.attrs = ldif_to_uldif(self.attrs)
self.attrs = ldif_to_cldif(self.attrs, conn, check_ctxt = True)
self._modifs = ldif_to_uldif(ldif)
self._modifs = ldif_to_cldif(self._modifs, conn, check_ctxt = False)
orig = ldif
elif dn != base_dn:
res = conn.search_s(dn, 0)
if not res:
raise ValueError ('objet inexistant: %s' % dn)
self.dn, self.attrs = res[0]
# Pour test en cas de mode w ou rw
orig = res[0][1]
self.attrs = ldif_to_uldif(self.attrs)
# L'objet sortant de la base ldap, on ne fait pas de vérifications sur
# l'état des données.
self.attrs = ldif_to_cldif(self.attrs, conn, check_ctxt = False)
self._modifs = ldif_to_cldif(ldif_to_uldif(res[0][1]), conn, check_ctxt = False)
# Je m'interroge sur la pertinence de cette partie, je pense qu'elle n'est
# pas utile. -- PEB 27/01/2013
if mode in ['w', 'rw']:
### Vérification que `λv. str(Attr(v))` est bien une projection
### C'est-à-dire que si on str(Attr(str(Attr(v)))) on retombe sur str(Attr(v))
oldif = orig
nldif = cldif_to_ldif(self.attrs)
for attr, vals in oldif.items():
if nldif[attr] != vals:
for v in nldif[attr]:
if v in vals:
vals.remove(v)
nvals = [nldif[attr][v.index(v)] for v in vals ]
raise EnvironmentError("λv. str(Attr(v)) n'est peut-être pas une projection (ie non idempotente):", attr, nvals, vals)
def _get_fields(self):
"""Renvoie la liste des champs LDAP de l'objet"""
return self.ofields + self.xfields + self.ufields + self.mfields
fields = property(_get_fields)
def history_add(self, login, chain):
"""Ajoute une ligne à l'historique de l'objet.
###ATTENTION : C'est un kludge pour pouvoir continuer à faire "comme avant",
### mais on devrait tout recoder pour utiliser l'historique LDAP"""
assert isinstance(login, str) or isinstance(login, unicode)
assert isinstance(chain, unicode)
new_line = u"%s, %s : %s" % (time.strftime("%d/%m/%Y %H:%M"), login, chain)
# Attention, le __setitem__ est surchargé, mais pas .append sur l'historique
self["historique"] = self["historique"] + [new_line]
def create(self):
"""Crée l'objet dans la base ldap, cette méthode vise à faire en sorte que
l'objet se crée lui-même, si celui qui essaye de le modifier a les droits
de le faire."""
# Création de la requête LDAP
modlist = addModlist(cldif_to_ldif(self.attrs))
# Requête LDAP de création de l'objet
self.add_s(self.dn, modlist)
def save(self):
"""Sauvegarde dans la base les modifications apportées à l'objet.
Interne: Vérifie que self._modifs contient des valeurs correctes et
enregistre les modifications."""
if self.mode not in ['w', 'rw']:
raise EnvironmentError("Objet en lecture seule, réessayer en lecture/écriture")
# On récupère la liste des modifications
modlist = self.get_modlist()
try:
self.conn.modify_s(self.dn, modlist)
except:
raise EnvironmentError("Impossible de modifier l'objet, peut-être n'existe-t-il pas ?")
# Vérification des modifications
self.attrs = ldif_to_uldif(self.conn.search_s(self.dn, 0)[0][1])
self.attrs = ldif_to_cldif(self.attrs, self.conn, check_ctxt=False)
differences = []
# On fait les différences entre les deux dicos
for attr in set(self.attrs.keys()).union(set(self._modifs.keys())):
exp_vals = set([unicode(i) for i in self.attrs.get(attr,[])])
new_vals = set([unicode(i) for i in self._modifs.get(attr,[])])
if exp_vals != new_vals:
differences.append({"missing": exp_vals - new_vals, "having": new_vals - exp_vals})
if differences:
raise EnvironmentError("Les modifications apportées à l'objet %s n'ont pas été correctement sauvegardées\n%s" % (self.dn, differences))
def may_be(self, what, obj):
"""Teste si celui qui est bindé peut effectuer ce qui est dans what, pour
what élément de {create, delete, modify}.
Retourne un booléen
"""
if set(obj.droits).intersection(self.can_be_by[what]) != set([]):
return True
else:
return False
def get_modlist(self):
"""Renvoie un dictionnaire des modifications apportées à l'objet"""
# unicode -> utf-8
ldif = cldif_to_ldif(self._modifs)
orig_ldif = cldif_to_ldif(self.attrs)
return modifyModlist(orig_ldif, ldif)
def get(self, attr, default):
"""Renvoie l'attribut demandé ou default si introuvable"""
try:
return self[attr]
except KeyError:
return default
def __getitem__(self, attr):
if self.mode in [ 'w', 'rw' ]:
return [ v for v in self._modifs[attr] ]
elif self.attrs.has_key(attr):
return [ v for v in self.attrs[attr] ]
elif self.has_key(attr):
return []
raise KeyError(attr)
def has_key(self,attr):
"""Est-ce que notre objet a l'attribut en question ?"""
return attr in self.ofields or attr in self.xfields or\
attr in self.ufields or attr in self.mfields
def __setitem__(self, attr, values):
if self.mode not in ['w', 'rw']:
raise ValueError("Objet en lecture seule")
if not isinstance(values, list):
values = [ values ]
self._modifs[attr] = [ attrify(val, attr, self._modifs, self.conn) for val in values ]
def search_historique(self, ign_fields=HIST_IGNORE_FIELDS):
u"""Récupère l'historique
l'argument optionnel ign_fields contient la liste des champs
à ignorer, HIST_IGNORE_FIELDS par défaut
Renvoie une liste de lignes de texte."""
res = self.conn.search_s(log_dn, ldap.SCOPE_SUBTREE, 'reqDN=%s' % self.dn)
res.sort(key=(lambda a: a[1]['reqEnd'][0]))
out = []
for cn, attrs in res:
date = crans_utils.format_ldap_time(attrs['reqEnd'][0])
author = attrs['reqAuthzID'][0]
if author == "cn=admin,dc=crans,dc=org":
author = u"respbats"
else:
author = author.split(",", 1)[0]
res = self.conn.search(author, scope=ldap.SCOPE_ONELEVEL)
if res != []:
author = res[0].compte()
if attrs['reqType'][0] == deleted:
out.append(u"%s : [%s] Suppression" % (date, author))
elif attrs['reqType'][0] == modified:
fields = {}
for mod in attrs['reqMod']:
mod = mod.decode('utf-8')
field, change = mod.split(':', 1)
if field not in ign_fields:
if field in fields:
fields[field].append(change)
else:
fields[field] = [change]
mod_list = []
for field in fields:
mods = fields[field]
mod_list.append(u"%s %s" %(field, ", ".join(mods)))
if mod_list != []:
out.append(u"%s : [%s] %s" % (date, author, u" ; ".join(mod_list)))
return out
def blacklist_actif(self):
"""Renvoie la liste des blacklistes actives sur l'entité
Améliorations possibles:
- Proposer de filtrer les blacklistes avec un arg supplémentaire ?
- Vérifier les blacklistes des machines pour les adhérents ?
"""
blacklist_liste=[]
# blacklistes virtuelle si on est un adhérent pour carte étudiant et chambre invalides
if self.__class__.__name__ == "adherent" and self.paiement_ok():
if not config.periode_transitoire and config.bl_carte_et_actif and not self.carte_ok() and not self.sursis_carte():
bl = blacklist(u'%s$%s$%s$%s' % ('-', '-', 'carte_etudiant', ''), {}, self.conn, False)
blacklist_liste.append(bl)
if self['chbre'][0].value == '????':
bl = blacklist(u'%s$%s$%s$%s' % ('-', '-', 'chambre_invalide', ''), {}, self.conn, False)
blacklist_liste.append(bl)
attrs = (self.attrs if self.mode not in ["w", "rw"] else self._modifs)
blacklist_liste.extend(filter((lambda bl: bl.is_actif()), attrs.get("blacklist",[])))
return blacklist_liste
def blacklist(self, sanction, commentaire, debut="now", fin = '-'):
u"""
Blacklistage de la ou de toutes la machines du propriétaire
* debut et fin sont le nombre de secondes depuis epoch
* pour un début ou fin immédiate mettre now
* pour une fin indéterminée mettre '-'
Les données sont stockées dans la base sous la forme :
debut$fin$sanction$commentaire
"""
if debut == 'now':
debut = int(time.time())
if fin == 'now':
fin = int(time.time())
bl = blacklist(u'%s$%s$%s$%s' % (debut, fin, sanction, commentaire), {}, self.conn, False)
self._modifs.setdefault('blacklist', []).append(bl)
class proprio(CransLdapObject):
u""" Un propriétaire de machine (adhérent, club…) """
can_be_by = { created: [nounou, bureau, cableur],
modified: [nounou, bureau, soi, cableur],
deleted: [nounou, bureau],
}
ufields = [ 'nom', 'chbre' ]
mfields = [ 'paiement', 'info', 'blacklist', 'controle']
ofields = []
xfields = []
def __init__(self, conn, dn, mode='ro', ldif = None, machines=[]):
super(proprio, self).__init__(conn, dn, mode, ldif)
self._machines = machines
def sursis_carte(self):
for h in self['historique'][::-1]:
x=re.match("(.*),.* : .*(paiement\+%s|inscription).*" % config.ann_scol,h.value)
if x != None:
return ((time.time()-time.mktime(time.strptime(x.group(1),'%d/%m/%Y %H:%M')))<=config.sursis_carte)
return False
def paiement_ok(self):
u"""Renvoie si le propriétaire a payé pour l'année en cours"""
if self.dn == base_dn:
return True
bool_paiement = False
try:
for paiement in self['paiement']:
if paiement.value == config.ann_scol:
bool_paiement = True
break
# Pour la période transitoire année précédente ok
if config.periode_transitoire and paiement.value == config.ann_scol -1:
bool_paiement = True
break
except KeyError:
pass
# Doit-on bloquer en cas de manque de la carte d'etudiant ?
# (si période transitoire on ne bloque dans aucun cas)
if config.bl_carte_et_definitif and not 'club' in map(lambda x:x.value,self["objectClass"]):
bool_carte = False
try:
for carte in self['carteEtudiant']:
if carte.value == config.ann_scol:
bool_carte = True
except KeyError:
pass
# Si inscrit depuis moins de config.sursis_carte, on laisse un sursis
if not bool_carte and self.sursis_carte():
bool_carte = True
return bool_carte and bool_paiement
return bool_paiement
def carte_ok(self):
u"""Renvoie si le propriétaire a donné sa carte pour l'année en cours"""
if not self.dn == base_dn and config.bl_carte_et_actif and not 'club' in map(lambda x:x.value,self["objectClass"]):
bool_carte = False
try:
for carte in self['carteEtudiant']:
if carte.value == config.ann_scol:
bool_carte = True
except KeyError:
pass
return bool_carte
return True
def update_solde(self, diff, comment=u"", login=None):
"""Modifie le solde du proprio. diff peut être négatif ou positif."""
login = login or current_user
assert isinstance(diff, int) or isinstance(diff, float)
assert isinstance(comment, unicode)
solde = float(self["solde"][0].value)
new_solde = solde + diff
# On vérifie qu'on ne dépasse par le découvert autorisé
if new_solde < config.impression.decouvert:
raise ValueError(u"Solde minimal atteint, opération non effectuée.")
transaction = u"credit" if diff >=0 else u"debit"
new_solde = u"%.2f" % new_solde
self.history_add(login, u"%s %.2f Euros [%s]" % (transaction, abs(diff), comment))
self["solde"] = new_solde
def machines(self):
"""Renvoie la liste des machines"""
if not self._machines:
self._machines = self.conn.search('mid=*', dn = self.dn, scope = 1)
for m in self._machines:
m._proprio = self
return self._machines
class machine(CransLdapObject):
u""" Une machine """
can_be_by = { created: [nounou, bureau, cableur, parent],
modified: [nounou, bureau, cableur, parent],
deleted: [nounou, bureau, cableur, parent],
}
ufields = ['mid', 'macAddress', 'host', 'midType']
ofields = ['rid']
mfields = ['info', 'blacklist', 'hostAlias', 'exempt',
'portTCPout', 'portTCPin', 'portUDPout', 'portUDPin','sshFingerprint', 'ipHostNumber', 'ip6HostNumber']
xfields = []
def __init__(self, conn, dn, mode='ro', ldif = None):
super(machine, self).__init__(conn, dn, mode, ldif)
self._proprio = None
def proprio(self):
u"""Renvoie le propriétaire de la machine"""
parent_dn = self.dn.split(',', 1)[1]
if not self._proprio:
self._proprio = new_cransldapobject(self.conn, parent_dn, self.mode)
return self._proprio
def blacklist_actif(self):
u"""Renvoie la liste des blacklistes actives sur la machine et le proprio
Améliorations possibles:
- Proposer de filtrer les blacklistes avec un arg supplémentaire ?
- Vérifier les blacklistes des machines pour les adhérents ?"""
black=self.proprio().blacklist_actif()
attrs = (self.attrs if self.mode not in ["w", "rw"] else self._modifs)
black.extend(filter((lambda bl: bl.is_actif()), attrs.get("blacklist",[])))
return black
class AssociationCrans(proprio):
u""" Association crans (propriétaire particulier)."""
pass
class adherent(proprio):
u"""Adhérent crans."""
ufields = proprio.ufields + ['aid', 'prenom', 'tel', 'mail', 'mailInvalide']
ofields = proprio.ofields + ['charteMA',
'canonicalAlias', 'solde', 'contourneGreylist',
'rewriteMailHeaders', 'derniereConnexion',
'homepageAlias','gpgFingerprint']
mfields = proprio.mfields + ['carteEtudiant', 'mailAlias', 'droits' ]
xfields = ['etudes', 'postalAddress']
def compte(self,login = None, uidNumber=0, hash_pass = '', shell=config.login_shell):
u"""Renvoie le nom du compte crans. S'il n'existe pas, et que uid
est précisé, le crée."""
if u'posixAccount' in self.attrs['objectClass']:
return self.attrs['uid'][0]
elif login:
fn = crans_utils.strip_accents(unicode(self.attrs['prenom'][0]).capitalize())
ln = crans_utils.strip_accents(unicode(self.attrs['nom'][0]).capitalize())
login = crans_utils.strip_accents(login).lower()
if jaro(ln.lower(), login) < 0.75 and jaro(fn.lower() + ' ' + ln.lower(), login) < 0.75:
raise ValueError("Le login est trop différent du nom",
login, self.attrs['nom'][0])
if not re.match('^[a-z][-a-z]{1,15}$', login):
raise ValueError("Le login a entre 2 et 16 lettres, il peut contenir (pas au début) des - ")
if crans_utils.mailexist(login):
raise ValueError("Login existant ou correspondant à un alias mail.")
home = u'/home/' + login
if os.path.exists(home):
raise ValueError('Création du compte impossible : home existant')
if os.path.exists("/var/mail/" + login):
raise ValueError('Création du compte impossible : /var/mail/%s existant' % login)
self._modifs['homeDirectory'] = [home]
self._modifs['mail'] = [login]
self._modifs['uid' ] = [login]
calias = crans_utils.strip_spaces(fn) + u'.' + crans_utils.strip_spaces(ln)
if crans_utils.mailexist(calias):
calias = login
self._modifs['canonicalAlias'] = [calias]
self._modifs['objectClass'] = [u'adherent', u'cransAccount', u'posixAccount', u'shadowAccount']
self._modifs['cn'] = [ fn + u' ' + ln ]
self._modifs['loginShell'] = [unicode(shell)]
self._modifs['userPassword'] = [unicode(hash_pass)]
if uidNumber:
if self.conn.search('(uidNumber=%s)' % uidNumber):
raise ValueError(u'uidNumber pris')
else:
pool_uid = range(1001, 9999)
random.shuffle(pool_uid)
while len(pool_uid) > 0:
uidNumber = pool_uid.pop() # On choisit un uid
if not self.conn.search('(uidNumber=%s)' % uidNumber):
break
if not len(pool_uid):
raise ValueError("Plus d'uid disponibles !")
## try:
## self.lock('uidNumber', str(uidNumber))
## except:
## # Quelqu'un nous a piqué l'uid que l'on venait de choisir !
## return self.compte(login, uidNumber, hash_pass, shell)
self._modifs['uidNumber'] = [unicode(uidNumber)]
self._modifs['gidNumber'] = [unicode(config.gid)]
self._modifs['gecos'] = [self._modifs['cn'][0] + u',,,']
self.save()
else:
raise EnvironmentError("L'adhérent n'a pas de compte crans")
class club(proprio):
u"""Club crans"""
ufields = ['cid', 'responsable']
mfields = ['imprimeurClub']
class machineFixe(machine):
u"""Machine fixe"""
pass
class machineWifi(machine):
u"""Machine wifi"""
ufields = machine.ufields + ['ipsec']
class machineCrans(machine):
can_be_by = { created: [nounou],
modified: [nounou],
deleted: [nounou],
}
ufields = machine.ufields + ['prise']
ofields = machine.ofields + ['nombrePrises']
class borneWifi(machine):
can_be_by = { created: [nounou],
modified: [nounou],
deleted: [nounou],
}
ufields = machine.ufields + ['canal', 'puissance', 'hotspot',
'prise', 'positionBorne', 'nvram']
class facture(CransLdapObject):
can_be_by = { created: [nounou, bureau, cableur],
modified: [nounou, bureau, cableur],
deleted: [nounou, bureau, cableur],
}
ufields = ['fid', 'modePaiement', 'recuPaiement']
class service(CransLdapObject): pass
class lock(CransLdapObject): pass
MODIFIABLE_ATTRS = [ 'tel', 'chbre', 'mailAlias', 'loginShell' ]
Reference in a new issue View git blame Copy permalink