network.py

# -*- coding: utf-8 -*-
"""
app.api.network
~~~~~~~~~~~~~~~

This module implements the network API.

:copyright: (c) 2017 European Spallation Source ERIC
:license: BSD 2-Clause, see LICENSE for more details.

"""
from flask import Blueprint, request
from flask_login import login_required, current_user
from .. import models
from ..decorators import login_groups_accepted
from ..utils import CSEntryError
from . import utils

bp = Blueprint("network_api", __name__)


@bp.route("/scopes")
@login_required
def get_scopes():
    """Return network scopes

    .. :quickref: Network; Get network scopes
    """
    return utils.get_generic_model(
        models.NetworkScope, order_by=models.NetworkScope.name
    )


@bp.route("/scopes", methods=["POST"])
@login_groups_accepted("admin")
def create_scope():
    """Create a new network scope

    .. :quickref: Network; Create new network scope

    :jsonparam name: network scope name
    :jsonparam first_vlan: network scope first vlan
    :jsonparam last_vlan: network scope last vlan
    :jsonparam supernet: network scope supernet
    :jsonparam domain_id: primary key of the default domain
    :jsonparam description: (optional) description
    """
    return utils.create_generic_model(
        models.NetworkScope,
        mandatory_fields=("name", "first_vlan", "last_vlan", "supernet", "domain_id"),
    )


@bp.route("/networks")
@login_required
def get_networks():
    """Return networks

    .. :quickref: Network; Get networks
    """
    return utils.get_generic_model(models.Network, order_by=models.Network.address)


@bp.route("/networks", methods=["POST"])
@login_groups_accepted("admin")
def create_network():
    """Create a new network

    .. :quickref: Network; Create new network

    :jsonparam vlan_name: vlan name
    :jsonparam vlan_id: vlan id
    :jsonparam address: vlan address
    :jsonparam first_ip: first IP of the allowed range
    :jsonparam last_ip: last IP of the allowed range
    :jsonparam gateway: gateway IP
    :jsonparam scope: network scope name
    :jsonparam domain_id: (optional) primary key of the domain [default: scope domain]
    :jsonparam admin_only: (optional) boolean to restrict the network to admin users [default: False]
    :type admin_only: bool
    :jsonparam description: (optional) description
    """
    return utils.create_generic_model(
        models.Network,
        mandatory_fields=(
            "vlan_name",
            "vlan_id",
            "address",
            "first_ip",
            "last_ip",
            "gateway",
            "scope",
        ),
    )


@bp.route("/interfaces")
@login_required
def get_interfaces():
    """Return interfaces

    .. :quickref: Network; Get interfaces
    """
    domain = request.args.get("domain", None)
    if domain is not None:
        query = models.Interface.query
        query = (
            query.join(models.Interface.network)
            .join(models.Network.domain)
            .filter(models.Domain.name == domain)
        )
        query = query.order_by(models.Interface.ip)
        return utils.get_generic_model(model=None, query=query)
    network = request.args.get("network", None)
    if network is not None:
        query = models.Interface.query
        query = query.join(models.Interface.network).filter(
            models.Network.vlan_name == network
        )
        query = query.order_by(models.Interface.ip)
        return utils.get_generic_model(model=None, query=query)
    return utils.get_generic_model(models.Interface, order_by=models.Interface.ip)


@bp.route("/interfaces", methods=["POST"])
@login_groups_accepted("admin", "network")
def create_interface():
    """Create a new interface

    .. :quickref: Network; Create new interface

    :jsonparam network: network name
    :jsonparam ip: interface IP
    :jsonparam name: interface name
    :jsonparam host: host name
    :jsonparam mac: (optional) MAC address
    """
    # The validate_interfaces method from the Network class is called when
    # setting interface.network. This is why we don't pass network_id here
    # but network (as vlan_name string)
    # Same for host
    data = request.get_json()
    # Check that the user has the permissions to create an interface on this network
    try:
        network = models.Network.query.filter_by(
            vlan_name=data["network"]
        ).first_or_404()
    except Exception:
        # If we can't get a network, an error will be raised in create_generic_model
        pass
    else:
        if not current_user.has_access_to_network(network):
            raise CSEntryError("User doesn't have the required group", status_code=403)
    return utils.create_generic_model(
        models.Interface, mandatory_fields=("network", "ip", "name", "host")
    )


@bp.route("/interfaces/<int:interface_id>", methods=["DELETE"])
@login_groups_accepted("admin")
def delete_interface(interface_id):
    """Delete an interface

    .. :quickref: Network; Delete an interface

    :param interface_id: interface primary key
    """
    return utils.delete_generic_model(models.Interface, interface_id)


@bp.route("/groups")
@login_required
def get_ansible_groups():
    """Return ansible groups

    .. :quickref: Network; Get Ansible groups
    """
    return utils.get_generic_model(
        models.AnsibleGroup, order_by=models.AnsibleGroup.name
    )


@bp.route("/groups", methods=["POST"])
@login_groups_accepted("admin")
def create_ansible_groups():
    """Create a new Ansible group

    .. :quickref: Network; Create new Ansible group

    :jsonparam name: group name
    :jsonparam vars: (optional) Ansible variables
    """
    return utils.create_generic_model(models.AnsibleGroup, mandatory_fields=("name",))


@bp.route("/hosts")
@login_required
def get_hosts():
    """Return hosts

    .. :quickref: Network; Get hosts
    """
    return utils.get_generic_model(models.Host, order_by=models.Host.name)


@bp.route("/hosts/search")
@login_required
def search_hosts():
    """Search hosts

    .. :quickref: Network; Search hosts

    :query q: the search query
    """
    return utils.search_generic_model(models.Host)


@bp.route("/hosts", methods=["POST"])
@login_groups_accepted("admin", "network")
def create_host():
    """Create a new host

    .. :quickref: Network; Create new host

    :jsonparam name: hostname
    :jsonparam device_type: Physical|Virtual|...
    :jsonparam is_ioc: True|False (optional)
    :jsonparam description: (optional) description
    :jsonparam items: (optional) list of items ICS id linked to the host
    :jsonparam ansible_vars: (optional) Ansible variables
    :jsonparam ansible_groups: (optional) list of Ansible groups names
    """
    return utils.create_generic_model(
        models.Host, mandatory_fields=("name", "device_type")
    )


@bp.route("/hosts/<int:host_id>", methods=["DELETE"])
@login_groups_accepted("admin")
def delete_host(host_id):
    """Delete a host

    .. :quickref: Network; Delete a host

    :param host_id: host primary key
    """
    return utils.delete_generic_model(models.Host, host_id)


@bp.route("/domains")
@login_required
def get_domains():
    """Return domains

    .. :quickref: Network; Get domains
    """
    return utils.get_generic_model(models.Domain, order_by=models.Domain.name)


@bp.route("/domains", methods=["POST"])
@login_groups_accepted("admin")
def create_domain():
    """Create a new domain

    .. :quickref: Network; Create new domain

    :jsonparam name: domain name
    """
    return utils.create_generic_model(models.Domain, mandatory_fields=("name",))


@bp.route("/cnames")
@login_required
def get_cnames():
    """Return cnames

    .. :quickref: Network; Get cnames
    """
    domain = request.args.get("domain", None)
    if domain is not None:
        query = models.Cname.query
        query = (
            query.join(models.Cname.interface)
            .join(models.Interface.network)
            .join(models.Network.domain)
            .filter(models.Domain.name == domain)
        )
        query = query.order_by(models.Cname.name)
        return utils.get_generic_model(model=None, query=query)
    return utils.get_generic_model(models.Cname, order_by=models.Cname.name)


@bp.route("/cnames", methods=["POST"])
@login_groups_accepted("admin")
def create_cname():
    """Create a new cname

    .. :quickref: Network; Create new cname

    :jsonparam name: full cname
    :jsonparam interface_id: primary key of the associated interface
    """
    return utils.create_generic_model(
        models.Cname, mandatory_fields=("name", "interface_id")
    )