1
0
mirror of https://github.com/GNS3/gns3-server synced 2024-12-04 22:28:12 +00:00
gns3-server/gns3server/api/routes/compute/images.py

160 lines
4.8 KiB
Python
Raw Normal View History

2020-10-02 06:37:50 +00:00
#
# Copyright (C) 2020 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""
API routes for images.
2020-10-02 06:37:50 +00:00
"""
import os
import urllib.parse
from fastapi import APIRouter, Request, status, Response, HTTPException
2020-10-02 06:37:50 +00:00
from fastapi.responses import FileResponse
from typing import List
from gns3server.compute.docker import Docker
from gns3server.compute.dynamips import Dynamips
from gns3server.compute.iou import IOU
from gns3server.compute.qemu import Qemu
router = APIRouter()
@router.get("/docker/images")
async def get_docker_images() -> List[str]:
"""
Get all Docker images.
"""
docker_manager = Docker.instance()
return await docker_manager.list_images()
@router.get("/dynamips/images")
async def get_dynamips_images() -> List[str]:
"""
Get all Dynamips images.
"""
dynamips_manager = Dynamips.instance()
return await dynamips_manager.list_images()
2021-04-13 09:16:50 +00:00
@router.post("/dynamips/images/{filename:path}", status_code=status.HTTP_204_NO_CONTENT)
async def upload_dynamips_image(filename: str, request: Request) -> Response:
2020-10-02 06:37:50 +00:00
"""
Upload a Dynamips IOS image.
"""
dynamips_manager = Dynamips.instance()
await dynamips_manager.write_image(urllib.parse.unquote(filename), request.stream())
return Response(status_code=status.HTTP_204_NO_CONTENT)
2020-10-02 06:37:50 +00:00
@router.get("/dynamips/images/{filename:path}")
async def download_dynamips_image(filename: str) -> FileResponse:
2020-10-02 06:37:50 +00:00
"""
Download a Dynamips IOS image.
"""
filename = urllib.parse.unquote(filename)
# Raise error if user try to escape
if filename[0] == "." or os.path.sep in filename:
2020-10-02 06:37:50 +00:00
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
dynamips_manager = Dynamips.instance()
image_path = dynamips_manager.get_abs_image_path(filename)
2020-10-02 06:37:50 +00:00
if not os.path.exists(image_path):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
return FileResponse(image_path, media_type="application/octet-stream")
@router.get("/iou/images")
async def get_iou_images() -> List[str]:
"""
Get all IOU images.
"""
iou_manager = IOU.instance()
return await iou_manager.list_images()
2021-04-13 09:16:50 +00:00
@router.post("/iou/images/{filename:path}", status_code=status.HTTP_204_NO_CONTENT)
async def upload_iou_image(filename: str, request: Request) -> Response:
2020-10-02 06:37:50 +00:00
"""
Upload an IOU image.
"""
iou_manager = IOU.instance()
await iou_manager.write_image(urllib.parse.unquote(filename), request.stream())
return Response(status_code=status.HTTP_204_NO_CONTENT)
2020-10-02 06:37:50 +00:00
@router.get("/iou/images/{filename:path}")
async def download_iou_image(filename: str) -> FileResponse:
2020-10-02 06:37:50 +00:00
"""
Download an IOU image.
"""
filename = urllib.parse.unquote(filename)
# Raise error if user try to escape
if filename[0] == "." or os.path.sep in filename:
2020-10-02 06:37:50 +00:00
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
iou_manager = IOU.instance()
image_path = iou_manager.get_abs_image_path(filename)
2020-10-02 06:37:50 +00:00
if not os.path.exists(image_path):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
return FileResponse(image_path, media_type="application/octet-stream")
@router.get("/qemu/images")
async def get_qemu_images() -> List[str]:
2020-10-02 06:37:50 +00:00
qemu_manager = Qemu.instance()
return await qemu_manager.list_images()
2021-04-13 09:16:50 +00:00
@router.post("/qemu/images/{filename:path}", status_code=status.HTTP_204_NO_CONTENT)
async def upload_qemu_image(filename: str, request: Request) -> Response:
2020-10-02 06:37:50 +00:00
qemu_manager = Qemu.instance()
await qemu_manager.write_image(urllib.parse.unquote(filename), request.stream())
return Response(status_code=status.HTTP_204_NO_CONTENT)
2020-10-02 06:37:50 +00:00
@router.get("/qemu/images/{filename:path}")
async def download_qemu_image(filename: str) -> FileResponse:
2020-10-02 06:37:50 +00:00
filename = urllib.parse.unquote(filename)
# Raise error if user try to escape
if filename[0] == "." or os.path.sep in filename:
2020-10-02 06:37:50 +00:00
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
qemu_manager = Qemu.instance()
2020-10-02 06:37:50 +00:00
image_path = qemu_manager.get_abs_image_path(filename)
if not os.path.exists(image_path):
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND)
return FileResponse(image_path, media_type="application/octet-stream")