copyparty vulnerable to reflected cross-site scripting via k304 parameter

@@ -337,6 +337,15 @@ def run(self) -> bool:
             vpath, arglist = self.req.split("?", 1)
             self.trailing_slash = vpath.endswith("/")
             vpath = undot(vpath)
+
+            zs = unquotep(arglist)
+            m = self.conn.hsrv.ptn_cc.search(zs)
+            if m:
+                hit = zs[m.span()[0] :]
+                t = "malicious user; Cc in query [{}] => [{!r}]"
+                self.log(t.format(self.req, hit), 1)
+                return False
+
             for k in arglist.split("&"):
                 if "=" in k:
                     k, zs = k.split("=", 1)
@@ -488,6 +497,9 @@ def run(self) -> bool:
                 pex: Pebkac = ex  # type: ignore
 
             try:
+                if pex.code == 999:
+                    return False
+
                 post = self.mode in ["POST", "PUT"] or "content-length" in self.headers
                 if not self._check_nonfatal(pex, post):
                     self.keepalive = False
@@ -586,6 +598,14 @@ def send_headers(
         for k, zs in list(self.out_headers.items()) + self.out_headerlist:
             response.append("%s: %s" % (k, zs))
 
+        for zs in response:
+            m = self.conn.hsrv.ptn_cc.search(zs)
+            if m:
+                hit = zs[m.span()[0] :]
+                t = "malicious user; Cc in out-hdr {!r} => [{!r}]"
+                self.log(t.format(zs, hit), 1)
+                raise Pebkac(999)
+
         try:
             # best practice to separate headers and body into different packets
             self.s.sendall("\r\n".join(response).encode("utf-8") + b"\r\n\r\n")
@@ -785,7 +805,7 @@ def handle_get(self) -> bool:
             path_base = os.path.join(self.E.mod, "web")
             static_path = absreal(os.path.join(path_base, self.vpath[5:]))
             if not static_path.startswith(path_base):
-                t = "attempted path traversal [{}] => [{}]"
+                t = "malicious user; attempted path traversal [{}] => [{}]"
                 self.log(t.format(self.vpath, static_path), 1)
                 self.tx_404()
                 return False
@@ -3077,7 +3097,14 @@ def tx_mounts(self) -> bool:
         return True
 
     def set_k304(self) -> bool:
-        ck = gencookie("k304", self.uparam["k304"], self.args.R, False, 86400 * 299)
+        v = self.uparam["k304"].lower()
+        if v == "y":
+            dur = 86400 * 299
+        else:
+            dur = None
+            v = "x"
+
+        ck = gencookie("k304", v, self.args.R, False, dur)
         self.out_headerlist.append(("Set-Cookie", ck))
         self.redirect("", "?h#cc")
         return True

@@ -4,6 +4,7 @@
 import base64
 import math
 import os
+import re
 import socket
 import sys
 import threading
@@ -138,6 +139,8 @@ def __init__(self, broker: "BrokerCli", nid: Optional[int]) -> None:
         zs = os.path.join(self.E.mod, "web", "deps", "prism.js.gz")
         self.prism = os.path.exists(zs)
 
+        self.ptn_cc = re.compile(r"[\x00-\x1f]")
+
         self.mallow = "GET HEAD POST PUT DELETE OPTIONS".split()
         if not self.args.no_dav:
             zs = "PROPFIND PROPPATCH LOCK UNLOCK MKCOL COPY MOVE"

@@ -171,6 +171,7 @@ def sunpack(fmt: bytes, a: bytes) -> tuple[Any, ...]:
     500: "Internal Server Error",
     501: "Not Implemented",
     503: "Service Unavailable",
+    999: "MissingNo",
 }