96 lines
3 KiB
Python
96 lines
3 KiB
Python
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",
|
|
)
|