summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorJohn Estabrook <jestabro@vyos.io>2021-06-12 11:43:29 -0500
committerJohn Estabrook <jestabro@vyos.io>2021-06-12 11:52:57 -0500
commit3a9041e2d4d4a48ba7c01439e69c5f86a4a850c2 (patch)
tree1b0399cd9f735c269ae2f69b6d1bec5b75889580
parent5cf4c64207b6526092e623476953044609b94c6c (diff)
downloadvyos-1x-3a9041e2d4d4a48ba7c01439e69c5f86a4a850c2.tar.gz
vyos-1x-3a9041e2d4d4a48ba7c01439e69c5f86a4a850c2.zip
http-api: T3616: update for strict content-type check in FastAPI 0.65.2
FastAPI 0.65.2 checks content-type request header before assuming JSON, closing a well-known loophole. This requires a modification of the code providing backwards compatibility of multipart forms.
-rwxr-xr-xsrc/services/vyos-http-api-server32
1 files changed, 32 insertions, 0 deletions
diff --git a/src/services/vyos-http-api-server b/src/services/vyos-http-api-server
index 8069d7146..cbf321dc8 100755
--- a/src/services/vyos-http-api-server
+++ b/src/services/vyos-http-api-server
@@ -32,6 +32,9 @@ from fastapi.responses import HTMLResponse
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
from pydantic import BaseModel, StrictStr, validator
+from starlette.datastructures import FormData, MutableHeaders
+from starlette.formparsers import FormParser, MultiPartParser
+from multipart.multipart import parse_options_header
import vyos.config
@@ -236,6 +239,35 @@ class MultipartRequest(Request):
ERR_PATH_NOT_LIST_OF_STR = False
offending_command = {}
exception = None
+
+ @property
+ def orig_headers(self):
+ self._orig_headers = super().headers
+ return self._orig_headers
+
+ @property
+ def headers(self):
+ self._headers = super().headers.mutablecopy()
+ self._headers['content-type'] = 'application/json'
+ return self._headers
+
+ async def form(self) -> FormData:
+ if not hasattr(self, "_form"):
+ assert (
+ parse_options_header is not None
+ ), "The `python-multipart` library must be installed to use form parsing."
+ content_type_header = self.orig_headers.get("Content-Type")
+ content_type, options = parse_options_header(content_type_header)
+ if content_type == b"multipart/form-data":
+ multipart_parser = MultiPartParser(self.orig_headers, self.stream())
+ self._form = await multipart_parser.parse()
+ elif content_type == b"application/x-www-form-urlencoded":
+ form_parser = FormParser(self.orig_headers, self.stream())
+ self._form = await form_parser.parse()
+ else:
+ self._form = FormData()
+ return self._form
+
async def body(self) -> bytes:
if not hasattr(self, "_body"):
forms = {}