coin/vpn/models.py

# -*- coding: utf-8 -*-
from __future__ import unicode_literals

from django.db import models
from django.core.exceptions import ValidationError
from django.conf import settings
from django.core.urlresolvers import reverse
from netfields import InetAddressField, NetManager
import ldapdb.models
from ldapdb.models.fields import CharField, ListField

from coin.mixins import CoinLdapSyncMixin
from coin.configuration.models import Configuration
# from coin.offers.backends import ValidateBackendType
from coin import utils
from coin import validation

"""BIG FAT WARNING

Ce code requiert une sévère factorisation avec housing/models.py et
vps/models.py

"""


class VPNConfiguration(CoinLdapSyncMixin, Configuration):
    url_namespace = "vpn"
    # backend_name = "openvpn_ldap"
    # administrative_subscription = models.OneToOneField(
    #     'offers.OfferSubscription',
    #     related_name=backend_name,
    #     validators=[ValidateBackendType(backend_name)])
    activated = models.BooleanField(default=False, verbose_name='activé')
    login = models.CharField(max_length=50, unique=True, blank=True,
                             verbose_name="identifiant",
                             help_text="Laisser vide pour une génération automatique")
    password = models.CharField(max_length=256, verbose_name="mot de passe",
                                blank=True, null=True)
    ipv4_endpoint = InetAddressField(validators=[validation.validate_v4],
                                     verbose_name="IPv4", blank=True, null=True,
                                     help_text="Adresse IPv4 utilisée par "
                                     "défaut sur le VPN")
    ipv6_endpoint = InetAddressField(validators=[validation.validate_v6],
                                     verbose_name="IPv6", blank=True, null=True,
                                     help_text="Adresse IPv6 utilisée par "
                                     "défaut sur le VPN")
    crypto_link = models.URLField(verbose_name="Matériel cryptographique", blank=True,
                           null=True, help_text="Lien à usage unique (détruit après ouverture)")
    objects = NetManager()

    def get_absolute_url(self):
        return reverse('vpn:details', args=[str(self.pk)])

    # This method is part of the general configuration interface.
    def subnet_event(self):
        self.check_endpoints(delete=True)
        # We potentially changed the endpoints, so we need to save.  Also,
        # saving will update the subnets in the LDAP backend.
        self.full_clean()
        self.save()

    def get_subnets(self, version):
        subnets = self.ip_subnet.all()
        return [subnet for subnet in subnets if subnet.inet.version == version]

    def sync_to_ldap(self, creation, *args, **kwargs):
        if creation:
            config = LdapVPNConfig()
        else:
            config = LdapVPNConfig.objects.get(pk=self.login)
        config.login = config.sn = self.login
        config.password = self.password
        config.active = 'yes' if self.activated else 'no'
        config.ipv4_endpoint = utils.str_or_none(self.ipv4_endpoint)
        config.ipv6_endpoint = utils.str_or_none(self.ipv6_endpoint)
        config.ranges_v4 = [str(s) for s in self.get_subnets(4)]
        config.ranges_v6 = [str(s) for s in self.get_subnets(6)]
        config.save()

    def delete_from_ldap(self):
        LdapVPNConfig.objects.get(pk=self.login).delete()

    def generate_endpoints(self, v4=True, v6=True):
        """Generate IP endpoints in one of the attributed IP subnets.  If there is
        no available subnet for a given address family, then no endpoint
        is generated for this address family.  If there already is an
        endpoint, do nothing.

        Returns True if an endpoint was generated.

        TODO: this should be factored for other technologies (DSL, etc)

        """
        subnets = self.ip_subnet.all()
        updated = False
        if v4 and self.ipv4_endpoint is None:
            subnets_v4 = [s for s in subnets if s.inet.version == 4]
            if len(subnets_v4) > 0:
                self.ipv4_endpoint = subnets_v4[0].inet.ip
                updated = True
        if v6 and self.ipv6_endpoint is None:
            subnets_v6 = [s for s in subnets if s.inet.version == 6]
            if len(subnets_v6) > 0:
                # With v6, we choose the second host of the subnet (cafe::1)
                gen = subnets_v6[0].inet.iter_hosts()
                gen.next()
                self.ipv6_endpoint = gen.next()
                updated = True
        return updated

    def check_endpoints(self, delete=False):
        """Check that the IP endpoints are included in one of the attributed IP
        subnets.

        If [delete] is True, then simply delete the faulty endpoints
        instead of raising an exception.
        """
        error = "L'IP {} n'est pas dans un réseau attribué."
        subnets = self.ip_subnet.all()
        is_faulty = lambda endpoint : endpoint and not any([endpoint in subnet.inet for subnet in subnets])
        if is_faulty(self.ipv4_endpoint):
            if delete:
                self.ipv4_endpoint = None
            else:
                raise ValidationError(error.format(self.ipv4_endpoint))
        if is_faulty(self.ipv6_endpoint):
            if delete:
                self.ipv6_endpoint = None
            else:
                raise ValidationError(error.format(self.ipv6_endpoint))

    def clean(self):
        # Generate VPN login, of the form "login-vpnX".  The resulting
        # login should not contain any ".", because graphite uses "." as a
        # separator.
        if not self.login:
            username = self.offersubscription.member.username
            vpns = VPNConfiguration.objects.filter(offersubscription__member__username=username)
            # This is the list of existing VPN logins for this user.
            logins = [vpn.login for vpn in vpns]
            # 100 VPNs ought to be enough for anybody.
            for login in ["{}-vpn{}".format(username, k) for k in range(1, 101)]:
                if login not in logins:
                    self.login = login
                    break
            # We may have failed.
            if not self.login:
                ValidationError("Impossible de générer un login VPN")
        # Hash password if needed
        self.password = utils.ldap_hash(self.password)
        # If saving for the first time and IP endpoints are not specified,
        # generate them automatically.
        if self.pk is None:
            self.generate_endpoints()
        self.check_endpoints()

    def __unicode__(self):
        return 'VPN ' + self.login

    class Meta:
        verbose_name = 'VPN'
        verbose_name_plural = 'VPN'


class LdapVPNConfig(ldapdb.models.Model):
    base_dn = settings.VPN_CONF_BASE_DN
    object_classes = [b'person', b'organizationalPerson', b'inetOrgPerson',
                      b'top', b'radiusprofile']

    login = CharField(db_column=b'cn', primary_key=True, max_length=255)
    sn = CharField(db_column=b'sn', max_length=255)
    password = CharField(db_column=b'userPassword', max_length=255)
    active = CharField(db_column=b'dialupAccess', max_length=3)
    ipv4_endpoint = CharField(db_column=b'radiusFramedIPAddress', max_length=16)
    ipv6_endpoint = CharField(db_column=b'postalAddress', max_length=40)
    ranges_v4 = ListField(db_column=b'radiusFramedRoute')
    ranges_v6 = ListField(db_column=b'registeredAddress')

    def __unicode__(self):
        return self.login

    class Meta:
        managed = False  # Indique à South de ne pas gérer le model LdapUser