diff --git a/gcp_storage_emulator/handlers/objects.py b/gcp_storage_emulator/handlers/objects.py
index 4e3ebc7..dccffc0 100644
--- a/gcp_storage_emulator/handlers/objects.py
+++ b/gcp_storage_emulator/handlers/objects.py
@@ -231,6 +231,28 @@ def _patch(storage, bucket_name, object_id, metadata):
         return None
 
 
+def xml_upload(request, response, storage, *args, **kwargs):
+    content_type = request.get_header("Content-Type", "application/octet-stream")
+    obj = _make_object_resource(
+        request.base_url,
+        request.params["bucket_name"],
+        request.params["object_id"],
+        content_type,
+        str(len(request.data)),
+    )
+    try:
+        obj = _checksums(request.data, obj)
+        storage.create_file(
+            request.params["bucket_name"],
+            request.params["object_id"],
+            request.data,
+            obj,
+        )
+
+    except NotFound:
+        response.status = HTTPStatus.NOT_FOUND
+
+
 def insert(request, response, storage, *args, **kwargs):
     uploadType = request.query.get("uploadType")
 
diff --git a/gcp_storage_emulator/server.py b/gcp_storage_emulator/server.py
index 6786ec2..3d9ff3b 100644
--- a/gcp_storage_emulator/server.py
+++ b/gcp_storage_emulator/server.py
@@ -85,8 +85,11 @@ def _health_check(req, res, storage):
     # Internal API, not supported by the real GCS
     (r"^/$", {GET: _health_check}),  # Health check endpoint
     (r"^/wipe$", {GET: _wipe_data}),  # Wipe all data
-    # Public file serving, same as object.download
-    (r"^/(?P<bucket_name>[-.\w]+)/(?P<object_id>.*[^/]+)$", {GET: objects.download}),
+    # Public file serving, same as object.download and signed URLs
+    (
+        r"^/(?P<bucket_name>[-.\w]+)/(?P<object_id>.*[^/]+)$",
+        {GET: objects.download, PUT: objects.xml_upload},
+    ),
 )
 
 BATCH_HANDLERS = (
@@ -161,15 +164,12 @@ def _decode_raw_data(raw_data, request_handler):
 
 
 def _read_data(request_handler):
-    if not request_handler.headers["Content-Type"]:
-        return None
-
     raw_data = _decode_raw_data(_read_raw_data(request_handler), request_handler)
 
     if not raw_data:
         return None
 
-    content_type = request_handler.headers["Content-Type"]
+    content_type = request_handler.headers["Content-Type"] or "application/octet-stream"
 
     if content_type.startswith("application/json"):
         return json.loads(raw_data)
diff --git a/tests/test_server.py b/tests/test_server.py
index c14b061..bd49310 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -7,11 +7,25 @@
 import fs
 import requests
 from google.api_core.exceptions import BadRequest, Conflict, NotFound
+from google.auth.credentials import AnonymousCredentials, Signing
 
 from gcp_storage_emulator.server import create_server
 from gcp_storage_emulator.settings import STORAGE_BASE, STORAGE_DIR
 
 
+class FakeSigningCredentials(Signing, AnonymousCredentials):
+    def sign_bytes(self, message):
+        return b"foobar"
+
+    @property
+    def signer_email(self):
+        return "foobar@example.tld"
+
+    @property
+    def signer(self):
+        pass
+
+
 def _get_storage_client(http):
     """Gets a python storage client"""
     os.environ["STORAGE_EMULATOR_HOST"] = "http://localhost:9023"
@@ -780,6 +794,49 @@ def test_empty_blob(self):
         fetched_content = blob.download_as_bytes()
         self.assertEqual(fetched_content, b"")
 
+    def test_signed_url_download(self):
+        content = b"The quick brown fox jumps over the lazy dog"
+        bucket = self._client.create_bucket("testbucket")
+
+        blob = bucket.blob("signed-download")
+        blob.upload_from_string(content)
+
+        url = blob.generate_signed_url(
+            api_access_endpoint="http://localhost:9023",
+            credentials=FakeSigningCredentials(),
+            version="v4",
+            expiration=datetime.timedelta(minutes=15),
+            method="GET",
+        )
+
+        response = requests.get(url)
+        self.assertEqual(response.content, content)
+
+    def test_signed_url_upload(self):
+        test_text = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)), "test_text.txt"
+        )
+        bucket = self._client.create_bucket("testbucket")
+
+        blob = bucket.blob("signed-upload")
+        url = blob.generate_signed_url(
+            api_access_endpoint="http://localhost:9023",
+            credentials=FakeSigningCredentials(),
+            version="v4",
+            expiration=datetime.timedelta(minutes=15),
+            method="PUT",
+        )
+
+        with open(test_text, "rb") as file:
+            headers = {"Content-type": "text/plain"}
+            response = requests.put(url, data=file, headers=headers)
+            self.assertEqual(response.status_code, 200)
+
+            blob_content = blob.download_as_bytes()
+            file.seek(0)
+            self.assertEqual(blob_content, file.read())
+            self.assertEqual(blob.content_type, "text/plain")
+
 
 class HttpEndpointsTest(ServerBaseCase):
     """Tests for the HTTP endpoints defined by server.HANDLERS."""