Fix wrong package stream resulting in 200 Ok

Assuming we want to keep our stream-redirect approach on the
content-app, We cant recover from wrong data already sent if
the Remote happens to be corrupted (contains wrong binaries).

In order to not give a 200 reponse to client, we decided to
close the connection as soon as the request handler realizes
the checksum is wrong.

That only happens after we already sent the whole blob minus EOF,
so we close the connection before sending the EOF.

fixes pulp#5012

CHANGES/5012.bugfix

@@ -0,0 +1,5 @@
+Fixed content-app behavior for the case where the client
+would get a 200 response for a package streamed from a Remote
+which didnt match the expected checksum.
+Now, the connection is closed right before finalizing the response.
+

pulpcore/content/handler.py

@@ -3,6 +3,8 @@
 from multidict import CIMultiDict
 import os
 import re
+import socket
+import struct
 from gettext import gettext as _
 
 from aiohttp.client_exceptions import ClientResponseError, ClientConnectionError
@@ -54,7 +56,10 @@
     cache_key,
 )
 
-from pulpcore.exceptions import UnsupportedDigestValidationError  # noqa: E402
+from pulpcore.exceptions import (  # noqa: E402
+    UnsupportedDigestValidationError,
+    DigestValidationError,
+)
 from pulpcore.metrics import artifacts_size_counter  # noqa: E402
 
 from jinja2 import Template  # noqa: E402: module level not at top of file
@@ -1125,13 +1130,27 @@ async def finalize():
                 await original_finalize()
 
         downloader = remote.get_downloader(
-            remote_artifact=remote_artifact, headers_ready_callback=handle_response_headers
+            remote_artifact=remote_artifact,
+            headers_ready_callback=handle_response_headers,
         )
         original_handle_data = downloader.handle_data
         downloader.handle_data = handle_data
         original_finalize = downloader.finalize
         downloader.finalize = finalize
-        download_result = await downloader.run()
+        try:
+            download_result = await downloader.run()
+        except DigestValidationError:
+            # Cant recover from wrong data already sent.
+            # We should close the connection without sending an EOF in the response
+            await downloader.session.close()
+            close_tcp_connection(request.transport._sock)
+            raise RuntimeError(
+                f"We tried streaming {remote_artifact.url!r} to the client, but it"
+                "failed checkusm validation. "
+                "At this point, we cant recover from wrong data already sent, "
+                "so we are forcing the connection to close. "
+                "If this error persists, the remote server might be corrupted."
+            )
 
         if content_length := response.headers.get("Content-Length"):
             response.headers["X-PULP-ARTIFACT-SIZE"] = content_length
@@ -1149,3 +1168,17 @@ async def finalize():
         if response.status == 404:
             raise HTTPNotFound()
         return response
+
+
+def close_tcp_connection(sock):
+    """Configure socket to close TCP connection immediately."""
+    try:
+        l_onoff = 1
+        l_linger = 0  # 0 seconds timeout - immediate close
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", l_onoff, l_linger))
+        # Another possibility is configure the socket to send a RST instead of FIN,
+        # but I'm not sure if that's required:
+        # https://serverfault.com/questions/242302/use-of-tcp-fin-and-tcp-rst
+        # sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    except (socket.error, OSError) as e:
+        log.warning(f"Error configuring socket for force close: {e}")