We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Server admin port still vulnerable to DOS
There are 128 connections allowed where each is limited to  512mb, this leaves 64GB of memory that can be acquired by a mal actor.

There is also still the issue where it is not checking if the socket is closed.

If I understand the idea of that port, the only data sent through it are some json files? I don't think it justifies such a large max size.

---

I think this is a larger problem though. Why is the the socket being accessed directly? There are many similar gotchas need to be considered when programming directly on a TCP socket and there are many libraries that have already done the hard work of solving those problems.

gRPC is an option since its already in your stack, Flask is an option but it doesn't match the use case too well, zeromq is an option

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nvflare/fuel/hci/server/hci.py`
Content:
```
1 # Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import logging
16 import socketserver
17 import ssl
18 import threading
19 
20 from nvflare.fuel.hci.conn import Connection, receive_til_end
21 from nvflare.fuel.hci.proto import validate_proto
22 from nvflare.fuel.hci.security import get_certificate_common_name
23 
24 from .reg import ServerCommandRegister
25 
26 MAX_ADMIN_CONNECTIONS = 128
27 
28 
29 class _MsgHandler(socketserver.BaseRequestHandler):
30     """Message handler.
31 
32     Used by the AdminServer to receive admin commands, validate, then process and do command through the
33     ServerCommandRegister.
34     """
35 
36     connections = 0
37     lock = threading.Lock()
38 
39     def __init__(self, request, client_address, server):
40         # handle() is called in the constructor so logger must be initialized first
41         self.logger = logging.getLogger(self.__class__.__name__)
42         super().__init__(request, client_address, server)
43 
44     def handle(self):
45         try:
46             with _MsgHandler.lock:
47                 _MsgHandler.connections += 1
48 
49             self.logger.debug(f"Concurrent admin connections: {_MsgHandler.connections}")
50             if _MsgHandler.connections > MAX_ADMIN_CONNECTIONS:
51                 raise ConnectionRefusedError(f"Admin connection limit ({MAX_ADMIN_CONNECTIONS}) reached")
52 
53             conn = Connection(self.request, self.server)
54 
55             if self.server.use_ssl:
56                 cn = get_certificate_common_name(self.request.getpeercert())
57                 conn.set_prop("_client_cn", cn)
58                 valid = self.server.validate_client_cn(cn)
59             else:
60                 valid = True
61 
62             if not valid:
63                 conn.append_error("authentication error")
64             else:
65                 req = receive_til_end(self.request).strip()
66                 command = None
67                 req_json = validate_proto(req)
68                 conn.request = req_json
69                 if req_json is not None:
70                     data = req_json["data"]
71                     for item in data:
72                         it = item["type"]
73                         if it == "command":
74                             command = item["data"]
75                             break
76 
77                     if command is None:
78                         conn.append_error("protocol violation")
79                     else:
80                         self.server.cmd_reg.process_command(conn, command)
81                 else:
82                     # not json encoded
83                     conn.append_error("protocol violation")
84 
85             if not conn.ended:
86                 conn.close()
87         except BaseException as exc:
88             self.logger.error(f"Admin connection terminated due to exception: {str(exc)}")
89             if self.logger.getEffectiveLevel() <= logging.DEBUG:
90                 self.logger.exception("Admin connection error")
91         finally:
92             with _MsgHandler.lock:
93                 _MsgHandler.connections -= 1
94 
95 
96 def initialize_hci():
97     socketserver.TCPServer.allow_reuse_address = True
98 
99 
100 class AdminServer(socketserver.ThreadingTCPServer):
101     # faster re-binding
102     allow_reuse_address = True
103 
104     # make this bigger than five
105     request_queue_size = 10
106 
107     # kick connections when we exit
108     daemon_threads = True
109 
110     def __init__(
111         self,
112         cmd_reg: ServerCommandRegister,
113         host,
114         port,
115         ca_cert=None,
116         server_cert=None,
117         server_key=None,
118         accepted_client_cns=None,
119     ):
120         """Base class of FedAdminServer to create a server that can receive commands.
121 
122         Args:
123             cmd_reg: CommandRegister
124             host: the IP address of the admin server
125             port: port number of admin server
126             ca_cert: the root CA's cert file name
127             server_cert: server's cert, signed by the CA
128             server_key: server's private key file
129             accepted_client_cns: list of accepted Common Names from client, if specified
130         """
131         socketserver.TCPServer.__init__(self, (host, port), _MsgHandler, False)
132 
133         self.use_ssl = False
134         if ca_cert and server_cert:
135             if accepted_client_cns:
136                 assert isinstance(accepted_client_cns, list), "accepted_client_cns must be list but got {}.".format(
137                     accepted_client_cns
138                 )
139 
140             ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
141             ctx.verify_mode = ssl.CERT_REQUIRED
142             ctx.load_verify_locations(ca_cert)
143             ctx.load_cert_chain(certfile=server_cert, keyfile=server_key)
144 
145             # replace the socket with an SSL version of itself
146             self.socket = ctx.wrap_socket(self.socket, server_side=True)
147             self.use_ssl = True
148 
149         # bind the socket and start the server
150         self.server_bind()
151         self.server_activate()
152 
153         self._thread = None
154         self.host = host
155         self.port = port
156         self.accepted_client_cns = accepted_client_cns
157         self.cmd_reg = cmd_reg
158         cmd_reg.finalize()
159         self.logger = logging.getLogger(self.__class__.__name__)
160 
161     def validate_client_cn(self, cn):
162         if self.accepted_client_cns:
163             return cn in self.accepted_client_cns
164         else:
165             return True
166 
167     def stop(self):
168         self.shutdown()
169         self.cmd_reg.close()
170 
171         if self._thread.is_alive():
172             self._thread.join()
173 
174         self.logger.info(f"Admin Server {self.host} on Port {self.port} shutdown!")
175 
176     def set_command_registry(self, cmd_reg: ServerCommandRegister):
177         if cmd_reg:
178             cmd_reg.finalize()
179 
180             if self.cmd_reg:
181                 self.cmd_reg.close()
182 
183             self.cmd_reg = cmd_reg
184 
185     def start(self):
186         if self._thread is None:
187             self._thread = threading.Thread(target=self._run, args=())
188 
189         if not self._thread.is_alive():
190             self._thread.start()
191 
192     def _run(self):
193         self.logger.info(f"Starting Admin Server {self.host} on Port {self.port}")
194         self.serve_forever()
195
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/nvflare/fuel/hci/server/hci.py b/nvflare/fuel/hci/server/hci.py
--- a/nvflare/fuel/hci/server/hci.py
+++ b/nvflare/fuel/hci/server/hci.py
@@ -23,7 +23,7 @@
 
 from .reg import ServerCommandRegister
 
-MAX_ADMIN_CONNECTIONS = 128
+MAX_ADMIN_CONNECTIONS = 16
 
 
 class _MsgHandler(socketserver.BaseRequestHandler):

{"golden_diff": "diff --git a/nvflare/fuel/hci/server/hci.py b/nvflare/fuel/hci/server/hci.py\n--- a/nvflare/fuel/hci/server/hci.py\n+++ b/nvflare/fuel/hci/server/hci.py\n@@ -23,7 +23,7 @@\n \n from .reg import ServerCommandRegister\n \n-MAX_ADMIN_CONNECTIONS = 128\n+MAX_ADMIN_CONNECTIONS = 16\n \n \n class _MsgHandler(socketserver.BaseRequestHandler):\n", "issue": "Server admin port still vulnerable to DOS\nThere are 128 connections allowed where each is limited to  512mb, this leaves 64GB of memory that can be acquired by a mal actor.\r\n\r\nThere is also still the issue where it is not checking if the socket is closed.\r\n\r\nIf I understand the idea of that port, the only data sent through it are some json files? I don't think it justifies such a large max size.\r\n\r\n---\r\n\r\nI think this is a larger problem though. Why is the the socket being accessed directly? There are many similar gotchas need to be considered when programming directly on a TCP socket and there are many libraries that have already done the hard work of solving those problems.\r\n\r\ngRPC is an option since its already in your stack, Flask is an option but it doesn't match the use case too well, zeromq is an option\n", "before_files": [{"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport logging\nimport socketserver\nimport ssl\nimport threading\n\nfrom nvflare.fuel.hci.conn import Connection, receive_til_end\nfrom nvflare.fuel.hci.proto import validate_proto\nfrom nvflare.fuel.hci.security import get_certificate_common_name\n\nfrom .reg import ServerCommandRegister\n\nMAX_ADMIN_CONNECTIONS = 128\n\n\nclass _MsgHandler(socketserver.BaseRequestHandler):\n    \"\"\"Message handler.\n\n    Used by the AdminServer to receive admin commands, validate, then process and do command through the\n    ServerCommandRegister.\n    \"\"\"\n\n    connections = 0\n    lock = threading.Lock()\n\n    def __init__(self, request, client_address, server):\n        # handle() is called in the constructor so logger must be initialized first\n        self.logger = logging.getLogger(self.__class__.__name__)\n        super().__init__(request, client_address, server)\n\n    def handle(self):\n        try:\n            with _MsgHandler.lock:\n                _MsgHandler.connections += 1\n\n            self.logger.debug(f\"Concurrent admin connections: {_MsgHandler.connections}\")\n            if _MsgHandler.connections > MAX_ADMIN_CONNECTIONS:\n                raise ConnectionRefusedError(f\"Admin connection limit ({MAX_ADMIN_CONNECTIONS}) reached\")\n\n            conn = Connection(self.request, self.server)\n\n            if self.server.use_ssl:\n                cn = get_certificate_common_name(self.request.getpeercert())\n                conn.set_prop(\"_client_cn\", cn)\n                valid = self.server.validate_client_cn(cn)\n            else:\n                valid = True\n\n            if not valid:\n                conn.append_error(\"authentication error\")\n            else:\n                req = receive_til_end(self.request).strip()\n                command = None\n                req_json = validate_proto(req)\n                conn.request = req_json\n                if req_json is not None:\n                    data = req_json[\"data\"]\n                    for item in data:\n                        it = item[\"type\"]\n                        if it == \"command\":\n                            command = item[\"data\"]\n                            break\n\n                    if command is None:\n                        conn.append_error(\"protocol violation\")\n                    else:\n                        self.server.cmd_reg.process_command(conn, command)\n                else:\n                    # not json encoded\n                    conn.append_error(\"protocol violation\")\n\n            if not conn.ended:\n                conn.close()\n        except BaseException as exc:\n            self.logger.error(f\"Admin connection terminated due to exception: {str(exc)}\")\n            if self.logger.getEffectiveLevel() <= logging.DEBUG:\n                self.logger.exception(\"Admin connection error\")\n        finally:\n            with _MsgHandler.lock:\n                _MsgHandler.connections -= 1\n\n\ndef initialize_hci():\n    socketserver.TCPServer.allow_reuse_address = True\n\n\nclass AdminServer(socketserver.ThreadingTCPServer):\n    # faster re-binding\n    allow_reuse_address = True\n\n    # make this bigger than five\n    request_queue_size = 10\n\n    # kick connections when we exit\n    daemon_threads = True\n\n    def __init__(\n        self,\n        cmd_reg: ServerCommandRegister,\n        host,\n        port,\n        ca_cert=None,\n        server_cert=None,\n        server_key=None,\n        accepted_client_cns=None,\n    ):\n        \"\"\"Base class of FedAdminServer to create a server that can receive commands.\n\n        Args:\n            cmd_reg: CommandRegister\n            host: the IP address of the admin server\n            port: port number of admin server\n            ca_cert: the root CA's cert file name\n            server_cert: server's cert, signed by the CA\n            server_key: server's private key file\n            accepted_client_cns: list of accepted Common Names from client, if specified\n        \"\"\"\n        socketserver.TCPServer.__init__(self, (host, port), _MsgHandler, False)\n\n        self.use_ssl = False\n        if ca_cert and server_cert:\n            if accepted_client_cns:\n                assert isinstance(accepted_client_cns, list), \"accepted_client_cns must be list but got {}.\".format(\n                    accepted_client_cns\n                )\n\n            ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)\n            ctx.verify_mode = ssl.CERT_REQUIRED\n            ctx.load_verify_locations(ca_cert)\n            ctx.load_cert_chain(certfile=server_cert, keyfile=server_key)\n\n            # replace the socket with an SSL version of itself\n            self.socket = ctx.wrap_socket(self.socket, server_side=True)\n            self.use_ssl = True\n\n        # bind the socket and start the server\n        self.server_bind()\n        self.server_activate()\n\n        self._thread = None\n        self.host = host\n        self.port = port\n        self.accepted_client_cns = accepted_client_cns\n        self.cmd_reg = cmd_reg\n        cmd_reg.finalize()\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n    def validate_client_cn(self, cn):\n        if self.accepted_client_cns:\n            return cn in self.accepted_client_cns\n        else:\n            return True\n\n    def stop(self):\n        self.shutdown()\n        self.cmd_reg.close()\n\n        if self._thread.is_alive():\n            self._thread.join()\n\n        self.logger.info(f\"Admin Server {self.host} on Port {self.port} shutdown!\")\n\n    def set_command_registry(self, cmd_reg: ServerCommandRegister):\n        if cmd_reg:\n            cmd_reg.finalize()\n\n            if self.cmd_reg:\n                self.cmd_reg.close()\n\n            self.cmd_reg = cmd_reg\n\n    def start(self):\n        if self._thread is None:\n            self._thread = threading.Thread(target=self._run, args=())\n\n        if not self._thread.is_alive():\n            self._thread.start()\n\n    def _run(self):\n        self.logger.info(f\"Starting Admin Server {self.host} on Port {self.port}\")\n        self.serve_forever()\n", "path": "nvflare/fuel/hci/server/hci.py"}], "after_files": [{"content": "# Copyright (c) 2021-2022, NVIDIA CORPORATION.  All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport logging\nimport socketserver\nimport ssl\nimport threading\n\nfrom nvflare.fuel.hci.conn import Connection, receive_til_end\nfrom nvflare.fuel.hci.proto import validate_proto\nfrom nvflare.fuel.hci.security import get_certificate_common_name\n\nfrom .reg import ServerCommandRegister\n\nMAX_ADMIN_CONNECTIONS = 16\n\n\nclass _MsgHandler(socketserver.BaseRequestHandler):\n    \"\"\"Message handler.\n\n    Used by the AdminServer to receive admin commands, validate, then process and do command through the\n    ServerCommandRegister.\n    \"\"\"\n\n    connections = 0\n    lock = threading.Lock()\n\n    def __init__(self, request, client_address, server):\n        # handle() is called in the constructor so logger must be initialized first\n        self.logger = logging.getLogger(self.__class__.__name__)\n        super().__init__(request, client_address, server)\n\n    def handle(self):\n        try:\n            with _MsgHandler.lock:\n                _MsgHandler.connections += 1\n\n            self.logger.debug(f\"Concurrent admin connections: {_MsgHandler.connections}\")\n            if _MsgHandler.connections > MAX_ADMIN_CONNECTIONS:\n                raise ConnectionRefusedError(f\"Admin connection limit ({MAX_ADMIN_CONNECTIONS}) reached\")\n\n            conn = Connection(self.request, self.server)\n\n            if self.server.use_ssl:\n                cn = get_certificate_common_name(self.request.getpeercert())\n                conn.set_prop(\"_client_cn\", cn)\n                valid = self.server.validate_client_cn(cn)\n            else:\n                valid = True\n\n            if not valid:\n                conn.append_error(\"authentication error\")\n            else:\n                req = receive_til_end(self.request).strip()\n                command = None\n                req_json = validate_proto(req)\n                conn.request = req_json\n                if req_json is not None:\n                    data = req_json[\"data\"]\n                    for item in data:\n                        it = item[\"type\"]\n                        if it == \"command\":\n                            command = item[\"data\"]\n                            break\n\n                    if command is None:\n                        conn.append_error(\"protocol violation\")\n                    else:\n                        self.server.cmd_reg.process_command(conn, command)\n                else:\n                    # not json encoded\n                    conn.append_error(\"protocol violation\")\n\n            if not conn.ended:\n                conn.close()\n        except BaseException as exc:\n            self.logger.error(f\"Admin connection terminated due to exception: {str(exc)}\")\n            if self.logger.getEffectiveLevel() <= logging.DEBUG:\n                self.logger.exception(\"Admin connection error\")\n        finally:\n            with _MsgHandler.lock:\n                _MsgHandler.connections -= 1\n\n\ndef initialize_hci():\n    socketserver.TCPServer.allow_reuse_address = True\n\n\nclass AdminServer(socketserver.ThreadingTCPServer):\n    # faster re-binding\n    allow_reuse_address = True\n\n    # make this bigger than five\n    request_queue_size = 10\n\n    # kick connections when we exit\n    daemon_threads = True\n\n    def __init__(\n        self,\n        cmd_reg: ServerCommandRegister,\n        host,\n        port,\n        ca_cert=None,\n        server_cert=None,\n        server_key=None,\n        accepted_client_cns=None,\n    ):\n        \"\"\"Base class of FedAdminServer to create a server that can receive commands.\n\n        Args:\n            cmd_reg: CommandRegister\n            host: the IP address of the admin server\n            port: port number of admin server\n            ca_cert: the root CA's cert file name\n            server_cert: server's cert, signed by the CA\n            server_key: server's private key file\n            accepted_client_cns: list of accepted Common Names from client, if specified\n        \"\"\"\n        socketserver.TCPServer.__init__(self, (host, port), _MsgHandler, False)\n\n        self.use_ssl = False\n        if ca_cert and server_cert:\n            if accepted_client_cns:\n                assert isinstance(accepted_client_cns, list), \"accepted_client_cns must be list but got {}.\".format(\n                    accepted_client_cns\n                )\n\n            ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)\n            ctx.verify_mode = ssl.CERT_REQUIRED\n            ctx.load_verify_locations(ca_cert)\n            ctx.load_cert_chain(certfile=server_cert, keyfile=server_key)\n\n            # replace the socket with an SSL version of itself\n            self.socket = ctx.wrap_socket(self.socket, server_side=True)\n            self.use_ssl = True\n\n        # bind the socket and start the server\n        self.server_bind()\n        self.server_activate()\n\n        self._thread = None\n        self.host = host\n        self.port = port\n        self.accepted_client_cns = accepted_client_cns\n        self.cmd_reg = cmd_reg\n        cmd_reg.finalize()\n        self.logger = logging.getLogger(self.__class__.__name__)\n\n    def validate_client_cn(self, cn):\n        if self.accepted_client_cns:\n            return cn in self.accepted_client_cns\n        else:\n            return True\n\n    def stop(self):\n        self.shutdown()\n        self.cmd_reg.close()\n\n        if self._thread.is_alive():\n            self._thread.join()\n\n        self.logger.info(f\"Admin Server {self.host} on Port {self.port} shutdown!\")\n\n    def set_command_registry(self, cmd_reg: ServerCommandRegister):\n        if cmd_reg:\n            cmd_reg.finalize()\n\n            if self.cmd_reg:\n                self.cmd_reg.close()\n\n            self.cmd_reg = cmd_reg\n\n    def start(self):\n        if self._thread is None:\n            self._thread = threading.Thread(target=self._run, args=())\n\n        if not self._thread.is_alive():\n            self._thread.start()\n\n    def _run(self):\n        self.logger.info(f\"Starting Admin Server {self.host} on Port {self.port}\")\n        self.serve_forever()\n", "path": "nvflare/fuel/hci/server/hci.py"}]}