You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
64 lines
1.7 KiB
64 lines
1.7 KiB
import logging
|
|
from typing import Optional
|
|
|
|
from flask import Response
|
|
from flask import blueprints
|
|
from flask import make_response
|
|
from flask import request
|
|
from flask_api import status
|
|
|
|
from .. import access
|
|
from .. import commands
|
|
from ..command_execution import execute_command
|
|
from ..commands import CommandId
|
|
from ..commands import CommandTarget
|
|
from ..commands import get_command_id_from_name
|
|
from ..commands import get_request_class
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
bp = blueprints.Blueprint("command", __name__)
|
|
|
|
|
|
@bp.route("/<role>/command", methods=["POST", "GET"])
|
|
@access.admin_permission.require(http_exception=status.HTTP_401_UNAUTHORIZED)
|
|
def command(role: str):
|
|
logger = _logger.getChild(f"{role}/command")
|
|
|
|
arguments = dict(request.args)
|
|
logger.debug(f"arguments: {arguments}")
|
|
|
|
cmd = arguments.pop("cmd")
|
|
target = CommandTarget[arguments.pop("target", role)]
|
|
|
|
try:
|
|
command_id = CommandId(int(cmd))
|
|
except ValueError:
|
|
command_id = get_command_id_from_name(cmd)
|
|
|
|
try:
|
|
command = get_request_class(command_id=command_id)(
|
|
root_logger=logging.getLogger(role), target=target, **arguments
|
|
)
|
|
except Exception:
|
|
return Response(status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
try:
|
|
response: Optional[commands.Response] = execute_command(
|
|
role=role,
|
|
command=command,
|
|
)
|
|
except KeyError:
|
|
logger.error(f"role {role} does not exist")
|
|
return Response(
|
|
f"role '{role}' does not exist",
|
|
status=status.HTTP_400_BAD_REQUEST,
|
|
)
|
|
|
|
if response is None:
|
|
return Response(status=status.HTTP_408_REQUEST_TIMEOUT)
|
|
|
|
return make_response(
|
|
response.as_dict(),
|
|
status.HTTP_200_OK,
|
|
)
|
|
|