coin/coin/vpn/models.py

# -*- coding: utf-8 -*-
from django.db import models
from django.core.exceptions import ValidationError
from netfields import InetAddressField, NetManager
import ldapdb.models
from ldapdb.models.fields import CharField, ListField

from coin.models import CoinLdapSyncModel
from coin.offers.models import OfferSubscription
from coin import utils


def validate_v4(address):
    if address.version != 4:
        raise ValidationError('{} is not an IPv4 address'.format(address))


def validate_v6(address):
    if address.version != 6:
        raise ValidationError('{} is not an IPv6 address'.format(address))


def str_or_none(obj):
    return str(obj) if obj else None


def validate_backend_type(subscription):
    """Ensures that the subscription we link to has the right backend
    type. This is a bit ugly, as validators are not meant for
    database-level sanity checks, but this is the cost of doing genericity
    the way we do.

    Note that this validator should not be needed for most cases, as the
    "limit_choices_to" parameter of the "administrative_subscription"
    OneToOneField automatically limits the available choices on forms.
    But it does not protect us if we fiddle manually with the database:
    better safe than sorry.

    """
    if OfferSubscription.objects.get(pk=subscription).offer.backend != 'openvpn_ldap':
        raise ValidationError('Administrative subscription must have a "openvpn_ldap" backend.')


class VPNSubscription(CoinLdapSyncModel):
    administrative_subscription = models.OneToOneField(
        'offers.OfferSubscription',
        related_name='openvpn_ldap',
        # Only consider VPN subscriptions.
        # TODO: also filter out subscriptions that already have a OneToOne
        # relation.  Simply adding "'openvpn_ldap': None" to the
        # dictionary does not work well, as it will also remove ourselves
        # from the list when editing the object in the admin.  Using a Q()
        # query does not solve the problem, as there is no way to
        # reference ourselves.  The proper solution would be to create a
        # custom form.
        limit_choices_to={'offer__backend': 'openvpn_ldap'},
        validators=[validate_backend_type])
    activated = models.BooleanField(default=False)
    login = models.CharField(max_length=50, unique=True)
    password = models.CharField(max_length=256)
    ipv4_endpoint = InetAddressField(validators=[validate_v4], blank=True,
                                     null=True)
    ipv6_endpoint = InetAddressField(validators=[validate_v6], blank=True,
                                     null=True)
    comment = models.CharField(blank=True, max_length=512)

    objects = NetManager()

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

    def sync_to_ldap(self, creation):
        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 = str_or_none(self.ipv4_endpoint)
        config.ipv6_endpoint = 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.administrative_subscription.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):
        """Check that the IP endpoints are included in one of the attributed IP
        subnets.
        """
        subnets = self.administrative_subscription.ip_subnet.all()
        for endpoint in [self.ipv4_endpoint, self.ipv6_endpoint]:
            if endpoint:
                if not any([endpoint in subnet.inet for subnet in subnets]):
                    raise ValidationError("Endpoint {} is not in an attributed range".format(endpoint))

    def clean(self):
        # 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 LdapVPNConfig(ldapdb.models.Model):
    # TODO: déplacer ligne suivante dans settings.py
    base_dn = "ou=vpn,o=ILLYSE,l=Villeurbanne,st=RHA,c=FR"
    object_classes = ['person', 'organizationalPerson', 'inetOrgPerson',
                      'top', 'radiusprofile']

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

    def __unicode__(self):
        return self.login

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