netbox/netbox/utilities/api.py

import logging

from django.contrib.contenttypes.fields import GenericForeignKey
from django.core.exceptions import (
    FieldDoesNotExist,
    FieldError,
    MultipleObjectsReturned,
    ObjectDoesNotExist,
    ValidationError,
)
from django.db.models.fields.related import ManyToOneRel, RelatedField
from django.urls import reverse
from django.utils.module_loading import import_string
from django.utils.translation import gettext_lazy as _
from rest_framework.permissions import BasePermission
from rest_framework.serializers import ListSerializer, Serializer
from rest_framework.views import get_view_name as drf_get_view_name

from extras.constants import HTTP_CONTENT_TYPE_JSON
from netbox.api.exceptions import GraphQLTypeNotFound, SerializerNotFound
from netbox.api.fields import RelatedObjectCountField
from netbox.registry import registry

from .query import count_related, dict_to_filter_params
from .string import title

logger = logging.getLogger('netbox.utilities.api')

__all__ = (
    'IsSuperuser',
    'get_annotations_for_serializer',
    'get_graphql_type_for_model',
    'get_prefetches_for_serializer',
    'get_related_object_by_attrs',
    'get_serializer_for_model',
    'get_view_name',
    'is_api_request',
    'is_graphql_request',
)


class IsSuperuser(BasePermission):
    """
    Allows access only to superusers.
    """
    def has_permission(self, request, view):
        return bool(request.user and request.user.is_active and request.user.is_superuser)


def get_serializer_for_model(model, prefix=''):
    """
    Return the appropriate REST API serializer for the given model.

    A plugin (or internal app) may register a custom resolver for its own
    app via netbox.plugins.register_serializer_resolver() to handle
    dynamically generated models or to override serializer resolution. If
    a resolver is registered for the model's app and returns a Serializer
    subclass, that result is used. Otherwise, the default import-path
    lookup runs.
    """
    app_label, model_name = model._meta.label.split('.')

    if resolver := registry['serializer_resolvers'].get(app_label):
        try:
            serializer = resolver(model, prefix=prefix)
        except Exception:
            # A buggy resolver must not break serializer lookup for the rest of NetBox.
            logger.exception("Serializer resolver %r raised an exception; falling through to default lookup.", resolver)
            serializer = None
        if serializer is not None:
            if isinstance(serializer, type) and issubclass(serializer, Serializer):
                return serializer
            logger.warning(
                "Serializer resolver %r returned %r, which is not a Serializer subclass; "
                "falling through to default lookup.",
                resolver, serializer,
            )

    serializer_name = f'{app_label}.api.serializers.{prefix}{model_name}Serializer'
    try:
        return import_string(serializer_name)
    except ImportError:
        raise SerializerNotFound(
            f"Could not determine serializer for {app_label}.{model_name} with prefix '{prefix}'"
        )


def get_graphql_type_for_model(model):
    """
    Return the GraphQL type class for the given model.
    """
    app_label, model_name = model._meta.label.split('.')
    class_name = f'{app_label}.graphql.types.{model_name}Type'
    try:
        return import_string(class_name)
    except ImportError:
        raise GraphQLTypeNotFound(f"Could not find GraphQL type for {app_label}.{model_name}")


def is_api_request(request):
    """
    Return True of the request is being made via the REST API.
    """
    return request.path_info.startswith(reverse('api-root'))


def is_graphql_request(request):
    """
    Return True of the request is being made via the GraphQL API.
    """
    return request.path_info == reverse('graphql') and request.content_type == HTTP_CONTENT_TYPE_JSON


def get_view_name(view):
    """
    Derive the view name from its associated model, if it has one. Fall back to DRF's built-in `get_view_name()`.
    This function is provided to DRF as its VIEW_NAME_FUNCTION.
    """
    if hasattr(view, 'queryset') and view.queryset is not None:
        # Derive the model name from the queryset.
        name = title(view.queryset.model._meta.verbose_name)
        if suffix := getattr(view, 'suffix', None):
            name = f'{name} {suffix}'
        return name

    # Fall back to DRF's default behavior
    return drf_get_view_name(view)


def _get_nested_serializer(serializer_field):
    """
    Return the nested serializer instance for a declared serializer field.
    """
    if isinstance(serializer_field, ListSerializer):
        serializer_field = serializer_field.child

    if isinstance(serializer_field, Serializer) and hasattr(serializer_field, 'nested'):
        return serializer_field

    return None


def _get_serializer_fields(serializer: Serializer):
    """
    Return the effective field names for a serializer instance, honoring any
    field-level fields=/omit= overrides.
    """
    fields = getattr(serializer, '_include_fields', None) or serializer.Meta.fields
    omit = getattr(serializer, '_omit_fields', []) or []

    return [field_name for field_name in fields if field_name not in omit]


def get_prefetches_for_serializer(serializer_class, fields=None, omit=None):
    """
    Compile and return a list of fields which should be prefetched on the queryset for a serializer.
    """
    if fields is not None and omit is not None:
        raise TypeError("Cannot specify both 'fields' and 'omit' parameters.")

    model = serializer_class.Meta.model

    # If fields are not specified, default to all
    fields_to_include = fields or serializer_class.Meta.fields
    fields_to_omit = omit or []

    prefetch_fields = []
    for field_name in fields_to_include:
        if field_name in fields_to_omit:
            continue
        serializer_field = serializer_class._declared_fields.get(field_name)

        # Determine the name of the model field referenced by the serializer field
        model_field_name = field_name
        if serializer_field and getattr(serializer_field, 'source', None):
            model_field_name = serializer_field.source

        # If the serializer field does not map to a discrete model field, skip it.
        try:
            field = model._meta.get_field(model_field_name)
            if isinstance(field, (RelatedField, ManyToOneRel, GenericForeignKey)):
                prefetch_fields.append(field.name)
        except FieldDoesNotExist:
            continue

        # If this field is represented by a nested serializer, recurse to resolve
        # prefetches for the related object, honoring any field-level fields=/omit=
        # constraints set on that serializer field instance.
        if nested_serializer := _get_nested_serializer(serializer_field):
            subfields = _get_serializer_fields(nested_serializer)
            for subfield in get_prefetches_for_serializer(type(nested_serializer), fields=subfields):
                prefetch_fields.append(f'{field.name}__{subfield}')

    return prefetch_fields


def get_annotations_for_serializer(serializer_class, fields=None, omit=None):
    """
    Return a mapping of field names to annotations to be applied to the queryset for a serializer.
    """
    if fields is not None and omit is not None:
        raise TypeError("Cannot specify both 'fields' and 'omit' parameters.")

    model = serializer_class.Meta.model

    # If fields are not specified, default to all
    fields_to_include = fields or serializer_class.Meta.fields
    fields_to_omit = omit or []

    annotations = {}
    for field_name, field in serializer_class._declared_fields.items():
        if field_name in fields_to_omit:
            continue
        if field_name in fields_to_include and type(field) is RelatedObjectCountField:
            related_field = getattr(model, field.relation).field
            annotations[field_name] = count_related(related_field.model, related_field.name)

    return annotations


def get_related_object_by_attrs(queryset, attrs, user=None):
    """
    Return an object identified by either a dictionary of attributes or its numeric primary key (ID). This is used
    for referencing related objects when creating/updating objects via the REST API.

    When a dictionary of attributes is provided, the queryset is first restricted to only those objects on which the
    given user has been granted view permission. This prevents an unprivileged user from enumerating objects by their
    attributes. Referencing an object directly by its numeric ID is always permitted, regardless of the user's view
    permissions.

    :param queryset: The base queryset from which to retrieve the related object
    :param attrs: A dictionary of attributes or a numeric primary key identifying the related object
    :param user: The user making the request (used to enforce view permissions on attribute-based lookups)
    """
    if attrs is None:
        return None

    # Dictionary of related object attributes
    if isinstance(attrs, dict):
        # Restrict the queryset to only those objects the user is permitted to view. This ensures that filtering by
        # attributes cannot be used to enumerate objects which the user is not otherwise permitted to see. Referencing
        # an object solely by its numeric ID (e.g. {"id": 123}) is equivalent to passing the ID directly, and is
        # always permitted regardless of the user's view permissions.
        if list(attrs) != ['id'] and user is not None and hasattr(queryset, 'restrict'):
            queryset = queryset.restrict(user, 'view')
        params = dict_to_filter_params(attrs)
        try:
            return queryset.get(**params)
        except ObjectDoesNotExist:
            raise ValidationError(
                _("Related object not found using the provided attributes: {params}").format(params=params))
        except MultipleObjectsReturned:
            raise ValidationError(
                _("Multiple objects match the provided attributes: {params}").format(params=params)
            )
        except FieldError as e:
            raise ValidationError(e)

    # Integer PK of related object
    try:
        # Cast as integer in case a PK was mistakenly sent as a string
        pk = int(attrs)
    except (TypeError, ValueError):
        raise ValidationError(
            _(
                "Related objects must be referenced by numeric ID or by dictionary of attributes. Received an "
                "unrecognized value: {value}"
            ).format(value=attrs)
        )

    # Look up object by PK
    try:
        return queryset.get(pk=pk)
    except ObjectDoesNotExist:
        raise ValidationError(_("Related object not found using the provided numeric ID: {id}").format(id=pk))