tasks.py

import logging

import requests
from celery import shared_task
from celery.exceptions import SoftTimeLimitExceeded
from django.conf import settings
from django.core.exceptions import ObjectDoesNotExist
from swapper import load_model

from openwisp_utils.tasks import OpenwispCeleryTask

from .utils import handle_error_notification, handle_recovery_notification

logger = logging.getLogger(__name__)


@shared_task(soft_time_limit=7200)
def update_template_related_config_status(template_pk):
    """
    Flags config objects related to the specified
    template PK as modified and triggers config
    modified and config status changed signals
    """
    Template = load_model("config", "Template")
    try:
        template = Template.objects.get(pk=template_pk)
    except ObjectDoesNotExist as e:
        logger.warning(
            f'update_template_related_config_status("{template_pk}") failed: {e}'
        )
        return
    try:
        template._update_related_config_status()
    except SoftTimeLimitExceeded:
        logger.error(
            "soft time limit hit while executing "
            f"_update_related_config_status for {template} "
            f"(ID: {template_pk})"
        )


@shared_task(soft_time_limit=7200)
def auto_add_template_to_existing_configs(template_pk):
    Template = load_model("config", "Template")
    try:
        template = Template.objects.get(pk=template_pk)
    except ObjectDoesNotExist as e:
        logger.warning(
            f'auto_add_template_to_existing_configs("{template_pk}") failed: {e}'
        )
        return
    try:
        template._auto_add_to_existing_configs()
    except SoftTimeLimitExceeded:
        logger.error(
            "soft time limit hit while executing "
            f"_auto_add_to_existing_configs for {template} "
            f"(ID: {template_pk})"
        )


@shared_task(soft_time_limit=1200)
def create_vpn_dh(vpn_pk):
    """
    Generates DH parameters
    """
    Vpn = load_model("config", "Vpn")
    vpn = Vpn.objects.get(pk=vpn_pk)
    try:
        vpn.dh = Vpn.dhparam(2048)
    except SoftTimeLimitExceeded:
        logger.error(
            "soft time limit hit while generating DH "
            f"parameters for VPN Server {vpn} (ID: {vpn_pk})"
        )
    else:
        vpn.full_clean()
        vpn.save()


@shared_task(soft_time_limit=7200)
def invalidate_devicegroup_cache_change(instance_id, model_name):
    from .api.views import DeviceGroupCommonName

    Device = load_model("config", "Device")
    DeviceGroup = load_model("config", "DeviceGroup")
    Cert = load_model("django_x509", "Cert")

    if model_name == Device._meta.model_name:
        DeviceGroupCommonName.device_change_invalidates_cache(instance_id)
    elif model_name == DeviceGroup._meta.model_name:
        DeviceGroupCommonName.devicegroup_change_invalidates_cache(instance_id)
    elif model_name == Cert._meta.model_name:
        DeviceGroupCommonName.certificate_change_invalidates_cache(instance_id)


@shared_task(soft_time_limit=7200)
def invalidate_vpn_server_devices_cache_change(vpn_pk):
    Vpn = load_model("config", "Vpn")
    VpnClient = load_model("config", "VpnClient")
    vpn = Vpn.objects.get(pk=vpn_pk)
    VpnClient.invalidate_clients_cache(vpn)


@shared_task(soft_time_limit=7200)
def invalidate_devicegroup_cache_delete(instance_id, model_name, **kwargs):
    from .api.views import DeviceGroupCommonName

    DeviceGroup = load_model("config", "DeviceGroup")
    Cert = load_model("django_x509", "Cert")

    if model_name == DeviceGroup._meta.model_name:
        DeviceGroupCommonName.devicegroup_delete_invalidates_cache(
            kwargs["organization_id"]
        )
    elif model_name == Cert._meta.model_name:
        DeviceGroupCommonName.certificate_delete_invalidates_cache(
            kwargs["organization_id"], kwargs["common_name"]
        )


# Generating large configurations can be time-consuming; hence,
# a custom soft time limit is applied here.
@shared_task(soft_time_limit=300)
def trigger_vpn_server_endpoint(endpoint, auth_token, vpn_id):
    Vpn = load_model("config", "Vpn")
    try:
        vpn = Vpn.objects.get(pk=vpn_id)
    except Vpn.DoesNotExist:
        logger.error(f"VPN Server UUID: {vpn_id} does not exist.")
        return

    # Cache the configuration here makes downloading the configuration faster.
    vpn.get_cached_configuration()
    task_key = f"vpn_update_task:{vpn_id}"
    try:
        response = requests.post(
            endpoint,
            params={"key": auth_token},
            verify=not settings.DEBUG,
            timeout=30,
        )
        response.raise_for_status()
    except requests.RequestException as e:
        logger.warning(
            f"Failed to update VPN Server configuration. "
            f"Error: {str(e)}, "
            f"VPN Server UUID: {vpn_id}"
        )
        handle_error_notification(
            task_key,
            sleep_time=5,
            exception=e,
            instance=vpn,
            action="update",
        )
    else:
        logger.info(f"Triggered update webhook of VPN Server UUID: {vpn_id}")
        handle_recovery_notification(
            task_key,
            sleep_time=5,
            instance=vpn,
            action="update",
        )


@shared_task(soft_time_limit=7200)
def change_devices_templates(instance_id, model_name, **kwargs):
    Device = load_model("config", "Device")
    DeviceGroup = load_model("config", "DeviceGroup")
    Config = load_model("config", "Config")
    if model_name == Device._meta.model_name:
        Device.manage_devices_group_templates(
            device_ids=instance_id,
            old_group_ids=kwargs.get("old_group_id"),
            group_id=kwargs.get("group_id"),
        )

    elif model_name == DeviceGroup._meta.model_name:
        DeviceGroup.manage_group_templates(
            group_id=instance_id,
            old_template_ids=kwargs.get("old_templates"),
            template_ids=kwargs.get("templates"),
        )

    elif model_name == Config._meta.model_name:
        Config.manage_backend_changed(
            instance_id=instance_id,
            old_backend=kwargs.pop("old_backend"),
            backend=kwargs.pop("backend"),
            **kwargs,
        )


@shared_task(soft_time_limit=7200)
def bulk_invalidate_config_get_cached_checksum(query_params):
    Config = load_model("config", "Config")
    Config.bulk_invalidate_get_cached_checksum(query_params)


@shared_task(base=OpenwispCeleryTask)
def invalidate_controller_views_cache(organization_id):
    """
    Invalidates the cache of DeviceChecksumView, GetVpnView
    """
    from .controller.views import DeviceChecksumView, GetVpnView

    Device = load_model("config", "Device")
    Vpn = load_model("config", "Vpn")

    for device in (
        Device.objects.filter(organization_id=organization_id).only("id").iterator()
    ):
        DeviceChecksumView.invalidate_get_device_cache(device)

    for vpn in (
        Vpn.objects.filter(organization_id=organization_id).only("id").iterator()
    ):
        GetVpnView.invalidate_get_vpn_cache(vpn)


@shared_task(base=OpenwispCeleryTask)
def invalidate_controller_views_for_group(group_id):
    """
    Invalidates the DeviceChecksumView cache only for devices in the given group.

    Unlike invalidate_controller_views_cache, this is scoped to a single device
    group and does not invalidate VPN caches.
    """
    from .controller.views import DeviceChecksumView

    Device = load_model("config", "Device")

    for device in Device.objects.filter(group_id=group_id).only("id").iterator():
        DeviceChecksumView.invalidate_get_device_cache(device)