Extract headers for login endpoint at param level

docs/index.html

@@ -1283,30 +1283,43 @@ <h1>PyNinja - Monitor<a class="headerlink" href="#pyninja-monitor" title="Permal
 <h1>Authenticator<a class="headerlink" href="#id1" title="Permalink to this heading">¶</a></h1>
 <span class="target" id="module-pyninja.monitor.authenticator"></span><dl class="py function">
 <dt class="sig sig-object py" id="pyninja.monitor.authenticator.failed_auth_counter">
-<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.authenticator.</span></span><span class="sig-name descname"><span class="pre">failed_auth_counter</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">request</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">Request</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">None</span></span></span><a class="headerlink" href="#pyninja.monitor.authenticator.failed_auth_counter" title="Permalink to this definition">¶</a></dt>
+<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.authenticator.</span></span><span class="sig-name descname"><span class="pre">failed_auth_counter</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">host</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">None</span></span></span><a class="headerlink" href="#pyninja.monitor.authenticator.failed_auth_counter" title="Permalink to this definition">¶</a></dt>
 <dd><p>Keeps track of failed login attempts from each host, and redirects if failed for 3 or more times.</p>
 <dl class="field-list simple">
 <dt class="field-odd">Parameters<span class="colon">:</span></dt>
-<dd class="field-odd"><p><strong>request</strong> – Takes the <code class="docutils literal notranslate"><span class="pre">Request</span></code> object as an argument.</p>
+<dd class="field-odd"><p><strong>host</strong> – Host header from the request.</p>
 </dd>
 </dl>
 </dd></dl>
 
 <dl class="py function">
 <dt class="sig sig-object py" id="pyninja.monitor.authenticator.raise_error">
-<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.authenticator.</span></span><span class="sig-name descname"><span class="pre">raise_error</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">request</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">NoReturn</span></span></span><a class="headerlink" href="#pyninja.monitor.authenticator.raise_error" title="Permalink to this definition">¶</a></dt>
+<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.authenticator.</span></span><span class="sig-name descname"><span class="pre">raise_error</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">host</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">NoReturn</span></span></span><a class="headerlink" href="#pyninja.monitor.authenticator.raise_error" title="Permalink to this definition">¶</a></dt>
 <dd><p>Raises a 401 Unauthorized error in case of bad credentials.</p>
+<dl class="field-list simple">
+<dt class="field-odd">Parameters<span class="colon">:</span></dt>
+<dd class="field-odd"><p><strong>host</strong> – Host header from the request.</p>
+</dd>
+</dl>
 </dd></dl>
 
 <dl class="py function">
 <dt class="sig sig-object py" id="pyninja.monitor.authenticator.extract_credentials">
-<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.authenticator.</span></span><span class="sig-name descname"><span class="pre">extract_credentials</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">request</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">Request</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">List</span><span class="p"><span class="pre">[</span></span><span class="pre">str</span><span class="p"><span class="pre">]</span></span></span></span><a class="headerlink" href="#pyninja.monitor.authenticator.extract_credentials" title="Permalink to this definition">¶</a></dt>
+<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.authenticator.</span></span><span class="sig-name descname"><span class="pre">extract_credentials</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">authorization</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span></span></em>, <em class="sig-param"><span class="n"><span class="pre">host</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">List</span><span class="p"><span class="pre">[</span></span><span class="pre">str</span><span class="p"><span class="pre">]</span></span></span></span><a class="headerlink" href="#pyninja.monitor.authenticator.extract_credentials" title="Permalink to this definition">¶</a></dt>
 <dd><p>Extract the credentials from <code class="docutils literal notranslate"><span class="pre">Authorization</span></code> headers and decode it before returning as a list of strings.</p>
+<dl class="field-list simple">
+<dt class="field-odd">Parameters<span class="colon">:</span></dt>
+<dd class="field-odd"><ul class="simple">
+<li><p><strong>authorization</strong> – Authorization header from the request.</p></li>
+<li><p><strong>host</strong> – Host header from the request.</p></li>
+</ul>
+</dd>
+</dl>
 </dd></dl>
 
 <dl class="py function">
 <dt class="sig sig-object py" id="pyninja.monitor.authenticator.verify_login">
-<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.authenticator.</span></span><span class="sig-name descname"><span class="pre">verify_login</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">request</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">Request</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">Dict</span><span class="p"><span class="pre">[</span></span><span class="pre">str</span><span class="p"><span class="pre">,</span></span><span class="w"> </span><span class="pre">Union</span><span class="p"><span class="pre">[</span></span><span class="pre">str</span><span class="p"><span class="pre">,</span></span><span class="w"> </span><span class="pre">int</span><span class="p"><span class="pre">]</span></span><span class="p"><span class="pre">]</span></span></span></span><a class="headerlink" href="#pyninja.monitor.authenticator.verify_login" title="Permalink to this definition">¶</a></dt>
+<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.authenticator.</span></span><span class="sig-name descname"><span class="pre">verify_login</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">authorization</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span></span></em>, <em class="sig-param"><span class="n"><span class="pre">host</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">Dict</span><span class="p"><span class="pre">[</span></span><span class="pre">str</span><span class="p"><span class="pre">,</span></span><span class="w"> </span><span class="pre">Union</span><span class="p"><span class="pre">[</span></span><span class="pre">str</span><span class="p"><span class="pre">,</span></span><span class="w"> </span><span class="pre">int</span><span class="p"><span class="pre">]</span></span><span class="p"><span class="pre">]</span></span></span></span><a class="headerlink" href="#pyninja.monitor.authenticator.verify_login" title="Permalink to this definition">¶</a></dt>
 <dd><p>Verifies authentication and generates session token for each user.</p>
 <dl class="field-list simple">
 <dt class="field-odd">Returns<span class="colon">:</span></dt>
@@ -1479,7 +1492,7 @@ <h1>Configuration<a class="headerlink" href="#configuration" title="Permalink to
 
 <dl class="py function">
 <dt class="sig sig-object py" id="pyninja.monitor.routes.login_endpoint">
-<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.routes.</span></span><span class="sig-name descname"><span class="pre">login_endpoint</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">request</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">Request</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">JSONResponse</span></span></span><a class="headerlink" href="#pyninja.monitor.routes.login_endpoint" title="Permalink to this definition">¶</a></dt>
+<em class="property"><span class="k"><span class="pre">async</span></span><span class="w"> </span></em><span class="sig-prename descclassname"><span class="pre">pyninja.monitor.routes.</span></span><span class="sig-name descname"><span class="pre">login_endpoint</span></span><span class="sig-paren">(</span><em class="sig-param"><span class="n"><span class="pre">request</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">Request</span></span></em>, <em class="sig-param"><span class="n"><span class="pre">authorization</span></span><span class="p"><span class="pre">:</span></span><span class="w"> </span><span class="n"><span class="pre">str</span></span><span class="w"> </span><span class="o"><span class="pre">=</span></span><span class="w"> </span><span class="default_value"><span class="pre">Header(None)</span></span></em><span class="sig-paren">)</span> <span class="sig-return"><span class="sig-return-icon">&#x2192;</span> <span class="sig-return-typehint"><span class="pre">JSONResponse</span></span></span><a class="headerlink" href="#pyninja.monitor.routes.login_endpoint" title="Permalink to this definition">¶</a></dt>
 <dd><p>Login endpoint for the monitoring page.</p>
 <dl class="field-list simple">
 <dt class="field-odd">Returns<span class="colon">:</span></dt>

pyninja/monitor/authenticator.py

@@ -13,26 +13,30 @@
 LOGGER = logging.getLogger("uvicorn.default")
 
 
-async def failed_auth_counter(request: Request) -> None:
+async def failed_auth_counter(host) -> None:
     """Keeps track of failed login attempts from each host, and redirects if failed for 3 or more times.
 
     Args:
-        request: Takes the ``Request`` object as an argument.
+        host: Host header from the request.
     """
     try:
-        models.ws_session.invalid[request.client.host] += 1
+        models.ws_session.invalid[host] += 1
     except KeyError:
-        models.ws_session.invalid[request.client.host] = 1
-    if models.ws_session.invalid[request.client.host] >= 3:
+        models.ws_session.invalid[host] = 1
+    if models.ws_session.invalid[host] >= 3:
         raise exceptions.RedirectException(location="/error")
 
 
-async def raise_error(request) -> NoReturn:
-    """Raises a 401 Unauthorized error in case of bad credentials."""
-    await failed_auth_counter(request)
+async def raise_error(host: str) -> NoReturn:
+    """Raises a 401 Unauthorized error in case of bad credentials.
+
+    Args:
+        host: Host header from the request.
+    """
+    await failed_auth_counter(host)
     LOGGER.error(
         "Incorrect username or password: %d",
-        models.ws_session.invalid[request.client.host],
+        models.ws_session.invalid[host],
     )
     raise exceptions.APIResponse(
         status_code=status.HTTP_401_UNAUTHORIZED,
@@ -41,42 +45,45 @@ async def raise_error(request) -> NoReturn:
     )
 
 
-async def extract_credentials(request: Request) -> List[str]:
-    """Extract the credentials from ``Authorization`` headers and decode it before returning as a list of strings."""
-    auth_header = request.headers.get("authorization", "")
-    # decode the Base64-encoded ASCII string
-    if not auth_header:
-        await raise_error(request)
-    decoded_auth = await monitor.secure.base64_decode(auth_header)
+async def extract_credentials(authorization: str, host: str) -> List[str]:
+    """Extract the credentials from ``Authorization`` headers and decode it before returning as a list of strings.
+
+    Args:
+        authorization: Authorization header from the request.
+        host: Host header from the request.
+    """
+    if not authorization:
+        await raise_error(host)
+    decoded_auth = await monitor.secure.base64_decode(authorization)
     # convert hex to a string
     auth = await monitor.secure.hex_decode(decoded_auth)
     return auth.split(",")
 
 
-async def verify_login(request: Request) -> Dict[str, Union[str, int]]:
+async def verify_login(authorization: str, host: str) -> Dict[str, Union[str, int]]:
     """Verifies authentication and generates session token for each user.
 
     Returns:
         Dict[str, str]:
         Returns a dictionary with the payload required to create the session token.
     """
-    username, signature, timestamp = await extract_credentials(request)
+    username, signature, timestamp = await extract_credentials(authorization, host)
     if secrets.compare_digest(username, models.env.monitor_username):
         hex_user = await monitor.secure.hex_encode(models.env.monitor_username)
         hex_pass = await monitor.secure.hex_encode(models.env.monitor_password)
     else:
         LOGGER.warning("User '%s' not allowed", username)
-        await raise_error(request)
+        await raise_error(host)
     message = f"{hex_user}{hex_pass}{timestamp}"
     expected_signature = await monitor.secure.calculate_hash(message)
     if secrets.compare_digest(signature, expected_signature):
-        models.ws_session.invalid[request.client.host] = 0
+        models.ws_session.invalid[host] = 0
         key = squire.keygen()
-        models.ws_session.client_auth[request.client.host] = dict(
+        models.ws_session.client_auth[host] = dict(
             username=username, token=key, timestamp=int(timestamp)
         )
-        return models.ws_session.client_auth[request.client.host]
-    await raise_error(request)
+        return models.ws_session.client_auth[host]
+    await raise_error(host)
 
 
 async def generate_cookie(auth_payload: dict) -> Dict[str, str | bool | int]:

pyninja/monitor/routes.py

@@ -3,7 +3,7 @@
 import time
 from http import HTTPStatus
 
-from fastapi import Cookie, Request
+from fastapi import Cookie, Header, Request
 from fastapi.responses import HTMLResponse, JSONResponse
 from fastapi.websockets import WebSocket, WebSocketDisconnect, WebSocketState
 
@@ -62,14 +62,18 @@ async def logout_endpoint(request: Request) -> HTMLResponse:
     return await monitor.config.clear_session(response)
 
 
-async def login_endpoint(request: Request) -> JSONResponse:
+async def login_endpoint(
+    request: Request, authorization: str = Header(None)
+) -> JSONResponse:
     """Login endpoint for the monitoring page.
 
     Returns:
         JSONResponse:
         Returns a JSONResponse object with a ``session_token`` and ``redirect_url`` set.
     """
-    auth_payload = await monitor.authenticator.verify_login(request)
+    auth_payload = await monitor.authenticator.verify_login(
+        authorization, request.client.host
+    )
     # AJAX calls follow redirect and return the response instead of replacing the URL
     # Solution is to revert to Form, but that won't allow header auth and additional customization done by JavaScript
     response = JSONResponse(

pyninja/monitor/templates/main.html

@@ -124,7 +124,7 @@ <h4>Swap Usage</h4>
                 data = JSON.parse(event.data);
             } catch (error) {
                 console.warn('Error parsing JSON data:', error);
-                alert(`${event.data}. Please refresh the page.`);
+                alert(event.data);
                 window.location.href = `${window.location.origin}/logout`;
                 return;
             }