fix: authorize the http connection to call commands (#2863)

fix: authorize the http connection to call DF commands

The assumption is that basic-auth already covers the authentication part.
And thanks to @sunneydev for finding the bug and providing the tests.
The tests actually uncovered another bug where we may parse partial http requests.
This one is handled by https://github.com/romange/helio/pull/243

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

helio

@@ -1 +1 @@
-Subproject commit 4f68b85a0b0a1d9220ae9b5a7d7d0c713dbae5bf
+Subproject commit f76c73fc6ca8cf1ada04edbb7e64a2465aa0f5b1

src/facade/dragonfly_connection.cc

@@ -659,6 +659,9 @@ void Connection::HandleRequests() {
       HttpConnection http_conn{http_listener_};
       http_conn.SetSocket(peer);
       http_conn.set_user_data(cc_.get());
+
+      // We validate the http request using basic-auth inside HttpConnection::HandleSingleRequest.
+      cc_->authenticated = true;
       auto ec = http_conn.ParseFromBuffer(io_buf_.InputBuffer());
       io_buf_.ConsumeInput(io_buf_.InputLen());
       if (!ec) {

src/server/http_api.cc

@@ -204,7 +204,9 @@ void HttpAPI(const http::QueryArgs& args, HttpRequest&& req, Service* service,
     }
   }
 
+  // TODO: to add a content-type/json check.
   if (!success) {
+    VLOG(1) << "Invalid body " << body;
     auto response = http::MakeStringResponse(h2::status::bad_request);
     http::SetMime(http::kTextMime, &response);
     response.body() = "Failed to parse json\r\n";

tests/dragonfly/connection_test.py

@@ -739,31 +739,3 @@ async def test_multiple_blocking_commands_client_pause(async_client: aioredis.Re
 
     assert not all.done()
     await all
-
-
-@dfly_args({"proactor_threads": "1", "expose_http_api": "true"})
-async def test_http(df_server: DflyInstance):
-    client = df_server.client()
-    async with ClientSession() as session:
-        async with session.get(f"http://localhost:{df_server.port}") as resp:
-            assert resp.status == 200
-
-        body = '["set", "foo", "МайяХилли", "ex", "100"]'
-        async with session.post(f"http://localhost:{df_server.port}/api", data=body) as resp:
-            assert resp.status == 200
-            text = await resp.text()
-            assert text.strip() == '{"result":"OK"}'
-
-        body = '["get", "foo"]'
-        async with session.post(f"http://localhost:{df_server.port}/api", data=body) as resp:
-            assert resp.status == 200
-            text = await resp.text()
-            assert text.strip() == '{"result":"МайяХилли"}'
-
-        body = '["foo", "bar"]'
-        async with session.post(f"http://localhost:{df_server.port}/api", data=body) as resp:
-            assert resp.status == 200
-            text = await resp.text()
-            assert text.strip() == '{"error": "unknown command `FOO`"}'
-
-    assert await client.ttl("foo") > 0

tests/dragonfly/http_conf_test.py

@@ -1,77 +1,129 @@
 import aiohttp
+from . import dfly_args
+from .instance import DflyInstance
 
 
-async def test_password(df_factory):
-    with df_factory.create(port=1112, requirepass="XXX") as server:
-        async with aiohttp.ClientSession() as session:
-            resp = await session.get(f"http://localhost:{server.port}/")
-            assert resp.status == 401
-        async with aiohttp.ClientSession(
-            auth=aiohttp.BasicAuth("default", "wrongpassword")
-        ) as session:
-            resp = await session.get(f"http://localhost:{server.port}/")
-            assert resp.status == 401
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("default", "XXX")) as session:
-            resp = await session.get(f"http://localhost:{server.port}/")
-            assert resp.status == 200
+def get_http_session(*args):
+    if args:
+        return aiohttp.ClientSession(auth=aiohttp.BasicAuth(*args))
+    return aiohttp.ClientSession()
 
 
-async def test_skip_metrics(df_factory):
-    with df_factory.create(port=1112, admin_port=1113, requirepass="XXX") as server:
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("whoops", "whoops")) as session:
-            resp = await session.get(f"http://localhost:{server.port}/metrics")
-            assert resp.status == 200
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("whoops", "whoops")) as session:
-            resp = await session.get(f"http://localhost:{server.admin_port}/metrics")
-            assert resp.status == 200
+@dfly_args({"proactor_threads": "1", "requirepass": "XXX"})
+async def test_password(df_server: DflyInstance):
+    async with get_http_session() as session:
+        resp = await session.get(f"http://localhost:{df_server.port}/")
+        assert resp.status == 401
+    async with get_http_session("default", "wrongpassword") as session:
+        resp = await session.get(f"http://localhost:{df_server.port}/")
+        assert resp.status == 401
+    async with get_http_session("default", "XXX") as session:
+        resp = await session.get(f"http://localhost:{df_server.port}/")
+        assert resp.status == 200
 
 
-async def test_no_password_main_port(df_factory):
-    with df_factory.create(
-        port=1112,
-    ) as server:
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("default", "XXX")) as session:
-            resp = await session.get(f"http://localhost:{server.port}/")
-            assert resp.status == 200
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("random")) as session:
-            resp = await session.get(f"http://localhost:{server.port}/")
-            assert resp.status == 200
-        async with aiohttp.ClientSession() as session:
-            resp = await session.get(f"http://localhost:{server.port}/")
-            assert resp.status == 200
+@dfly_args({"proactor_threads": "1", "requirepass": "XXX", "admin_port": 1113})
+async def test_skip_metrics(df_server: DflyInstance):
+    async with get_http_session("whoops", "whoops") as session:
+        resp = await session.get(f"http://localhost:{df_server.port}/metrics")
+        assert resp.status == 200
+    async with get_http_session("whoops", "whoops") as session:
+        resp = await session.get(f"http://localhost:{df_server.admin_port}/metrics")
+        assert resp.status == 200
 
 
-async def test_no_password_on_admin(df_factory):
-    with df_factory.create(
-        port=1112,
-        admin_port=1113,
-        requirepass="XXX",
-        primary_port_http_enabled=True,
-        admin_nopass=True,
-    ) as server:
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("default", "XXX")) as session:
-            resp = await session.get(f"http://localhost:{server.admin_port}/")
-            assert resp.status == 200
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("random")) as session:
-            resp = await session.get(f"http://localhost:{server.admin_port}/")
-            assert resp.status == 200
-        async with aiohttp.ClientSession() as session:
-            resp = await session.get(f"http://localhost:{server.admin_port}/")
-            assert resp.status == 200
+async def test_no_password_main_port(df_server: DflyInstance):
+    async with get_http_session("default", "XXX") as session:
+        resp = await session.get(f"http://localhost:{df_server.port}/")
+        assert resp.status == 200
+    async with get_http_session("random") as session:
+        resp = await session.get(f"http://localhost:{df_server.port}/")
+        assert resp.status == 200
+    async with get_http_session() as session:
+        resp = await session.get(f"http://localhost:{df_server.port}/")
+        assert resp.status == 200
 
 
-async def test_password_on_admin(df_factory):
-    with df_factory.create(
-        port=1112,
-        admin_port=1113,
-        requirepass="XXX",
-    ) as server:
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("default", "badpass")) as session:
-            resp = await session.get(f"http://localhost:{server.admin_port}/")
-            assert resp.status == 401
-        async with aiohttp.ClientSession() as session:
-            resp = await session.get(f"http://localhost:{server.admin_port}/")
-            assert resp.status == 401
-        async with aiohttp.ClientSession(auth=aiohttp.BasicAuth("default", "XXX")) as session:
-            resp = await session.get(f"http://localhost:{server.admin_port}/")
+@dfly_args(
+    {
+        "proactor_threads": "1",
+        "requirepass": "XXX",
+        "admin_port": 1113,
+        "primary_port_http_enabled": True,
+        "admin_nopass": True,
+    }
+)
+async def test_no_password_on_admin(df_server: DflyInstance):
+    async with get_http_session("default", "XXX") as session:
+        resp = await session.get(f"http://localhost:{df_server.admin_port}/")
+        assert resp.status == 200
+    async with get_http_session("random") as session:
+        resp = await session.get(f"http://localhost:{df_server.admin_port}/")
+        assert resp.status == 200
+    async with get_http_session() as session:
+        resp = await session.get(f"http://localhost:{df_server.admin_port}/")
+        assert resp.status == 200
+
+
+@dfly_args({"proactor_threads": "1", "requirepass": "XXX", "admin_port": 1113})
+async def test_password_on_admin(df_server: DflyInstance):
+    async with get_http_session("default", "badpass") as session:
+        resp = await session.get(f"http://localhost:{df_server.admin_port}/")
+        assert resp.status == 401
+    async with get_http_session() as session:
+        resp = await session.get(f"http://localhost:{df_server.admin_port}/")
+        assert resp.status == 401
+    async with get_http_session("default", "XXX") as session:
+        resp = await session.get(f"http://localhost:{df_server.admin_port}/")
+        assert resp.status == 200
+
+
+@dfly_args({"proactor_threads": "1", "expose_http_api": "true"})
+async def test_no_password_on_http_api(df_server: DflyInstance):
+    async with get_http_session("default", "XXX") as session:
+        resp = await session.post(f"http://localhost:{df_server.port}/api", json=["ping"])
+        assert resp.status == 200
+    async with get_http_session("random") as session:
+        resp = await session.post(f"http://localhost:{df_server.port}/api", json=["ping"])
+        assert resp.status == 200
+    async with get_http_session() as session:
+        resp = await session.post(f"http://localhost:{df_server.port}/api", json=["ping"])
+        assert resp.status == 200
+
+
+@dfly_args({"proactor_threads": "1", "expose_http_api": "true"})
+async def test_http_api(df_server: DflyInstance):
+    client = df_server.client()
+    async with get_http_session() as session:
+        body = '["set", "foo", "МайяХилли", "ex", "100"]'
+        async with session.post(f"http://localhost:{df_server.port}/api", data=body) as resp:
             assert resp.status == 200
+            text = await resp.text()
+            assert text.strip() == '{"result":"OK"}'
+
+        body = '["get", "foo"]'
+        async with session.post(f"http://localhost:{df_server.port}/api", data=body) as resp:
+            assert resp.status == 200
+            text = await resp.text()
+            assert text.strip() == '{"result":"МайяХилли"}'
+
+        body = '["foo", "bar"]'
+        async with session.post(f"http://localhost:{df_server.port}/api", data=body) as resp:
+            assert resp.status == 200
+            text = await resp.text()
+            assert text.strip() == '{"error": "unknown command `FOO`"}'
+
+    assert await client.ttl("foo") > 0
+
+
+@dfly_args({"proactor_threads": "1", "expose_http_api": "true", "requirepass": "XXX"})
+async def test_password_on_http_api(df_server: DflyInstance):
+    async with get_http_session("default", "badpass") as session:
+        resp = await session.post(f"http://localhost:{df_server.port}/api", json=["ping"])
+        assert resp.status == 401
+    async with get_http_session() as session:
+        resp = await session.post(f"http://localhost:{df_server.port}/api", json=["ping"])
+        assert resp.status == 401
+    async with get_http_session("default", "XXX") as session:
+        resp = await session.post(f"http://localhost:{df_server.port}/api", json=["ping"])
+        assert resp.status == 200