majuna/app/api/web.py

from datetime import datetime, timedelta, timezone
from typing import List

from flask import Blueprint, abort, request
from flask.typing import ResponseReturnValue

from app.api.util import (
    DOMAIN_NAME_REGEX,
    MAX_ALLOWED_ITEMS,
    MAX_DOMAIN_NAME_LENGTH,
    ListFilter,
    list_resources,
)
from app.models.base import Group
from app.models.mirrors import Origin, Proxy

api_web = Blueprint("web", __name__)


@api_web.route("/group", methods=["GET"])
def list_groups() -> ResponseReturnValue:
    return list_resources(
        Group,
        lambda group: group.to_dict(),
        resource_name="OriginGroupList",
        max_allowed_items=MAX_ALLOWED_ITEMS,
        protective_marking="amber",
    )


@api_web.route("/origin", methods=["GET"])
def list_origins() -> ResponseReturnValue:
    domain_name_filter = request.args.get("DomainName")
    group_id_filter = request.args.get("GroupId")

    filters: List[ListFilter] = []

    if domain_name_filter:
        if len(domain_name_filter) > MAX_DOMAIN_NAME_LENGTH:
            abort(
                400,
                description=f"DomainName cannot exceed {MAX_DOMAIN_NAME_LENGTH} characters.",
            )
        if not DOMAIN_NAME_REGEX.match(domain_name_filter):
            abort(400, description="DomainName contains invalid characters.")
        filters.append(Origin.domain_name.ilike(f"%{domain_name_filter}%"))

    if group_id_filter:
        try:
            filters.append(Origin.group_id == int(group_id_filter))
        except ValueError:
            abort(400, description="GroupId must be a valid integer.")

    return list_resources(
        Origin,
        lambda origin: origin.to_dict(),
        filters=filters,
        resource_name="OriginsList",
        max_allowed_items=MAX_ALLOWED_ITEMS,
        protective_marking="amber",
    )


@api_web.route("/mirror", methods=["GET"])
def list_mirrors() -> ResponseReturnValue:
    filters = []

    twenty_four_hours_ago = datetime.now(timezone.utc) - timedelta(hours=24)
    status_filter = request.args.get("Status")
    if status_filter:
        if status_filter == "pending":
            filters.append(Proxy.url.is_(None))
            filters.append(Proxy.deprecated.is_(None))
            filters.append(Proxy.destroyed.is_(None))
        if status_filter == "active":
            filters.append(Proxy.url.is_not(None))
            filters.append(Proxy.deprecated.is_(None))
            filters.append(Proxy.destroyed.is_(None))
        if status_filter == "expiring":
            filters.append(Proxy.deprecated.is_not(None))
            filters.append(Proxy.destroyed.is_(None))
        if status_filter == "destroyed":
            filters.append(Proxy.destroyed > twenty_four_hours_ago)
    else:
        filters.append(
            (Proxy.destroyed.is_(None)) | (Proxy.destroyed > twenty_four_hours_ago)
        )

    return list_resources(
        Proxy,
        lambda proxy: proxy.to_dict(),
        filters=filters,
        resource_name="MirrorsList",
        max_allowed_items=MAX_ALLOWED_ITEMS,
        protective_marking="amber",
    )