added firewall

Author: freelancing-solutions

requirements.txt

@@ -82,6 +82,7 @@ class Config:
 class CloudFlareSettings(BaseSettings):
     EMAIL: str = Field(..., env="CLOUDFLARE_EMAIL")
     TOKEN: str = Field(..., env="CLOUDFLARE_TOKEN")
+    X_CLIENT_SECRET_TOKEN: str = Field(..., env="X_CLIENT_SECRET_TOKEN")
 
     class Config:
         case_sensitive = True

src/config/config.py

@@ -1,6 +1,8 @@
 import re
+import hmac
 import ipaddress
-import httpx
+
+import requests
 from CloudFlare import CloudFlare
 from CloudFlare.exceptions import CloudFlareAPIError
 from flask import Flask, request
@@ -68,27 +70,6 @@ def contains_malicious_patterns(_input: str) -> bool:
 EMAIL: str = config_instance().CLOUDFLARE_SETTINGS.EMAIL
 TOKEN: str = config_instance().CLOUDFLARE_SETTINGS.TOKEN
 
-async_client = httpx.AsyncClient(http2=True, limits=httpx.Limits(max_connections=100, max_keepalive_connections=20))
-
-
-async def send_request(api_url: str, headers: dict[str, str | int], method: str = 'get',
-                       data: str | None = None):
-    try:
-        if method.lower() == "get":
-            response = await async_client.get(url=api_url, headers=headers, timeout=360000)
-        elif method.lower() == "post":
-            if data:
-                response = await async_client.post(url=api_url, json=data, headers=headers, timeout=360000)
-            else:
-                response = await async_client.post(url=api_url, headers=headers, timeout=360000)
-        else:
-            return None
-    except httpx.HTTPError as http_err:
-        raise http_err
-    except Exception as err:
-        raise err
-    return response.json()
-
 
 class Firewall:
     """
@@ -116,6 +97,7 @@ def init_app(self, app: Flask):
             app.before_request(self.is_host_valid)
             app.before_request(self.is_edge_ip_allowed)
             app.before_request(self.check_if_request_malicious)
+            app.before_request(self.verify_client_secret_token)
         # obtain the latest cloudflare edge servers
         ipv4, ipv6 = self.get_ip_ranges()
         # updating the ip ranges
@@ -161,6 +143,19 @@ def check_if_request_malicious(self):
         if any((pattern.match(path) for pattern in self.compiled_bad_patterns)):
             raise UnAuthenticatedError('Payload is suspicious')
 
+    @staticmethod
+    def verify_client_secret_token():
+        client_secret_token = request.headers.get('X_CLIENT_SECRET_TOKEN')
+        if not client_secret_token:
+            raise UnAuthenticatedError('Missing client secret token')
+
+        expected_secret_token = config_instance().CLOUDFLARE_SETTINGS.get('X_CLIENT_SECRET_TOKEN')
+        if not expected_secret_token:
+            raise ValueError('Missing expected client secret token')
+
+        if not hmac.compare_digest(client_secret_token, expected_secret_token):
+            raise UnAuthenticatedError('Invalid client secret token')
+
     @staticmethod
     def get_client_ip() -> str:
         """
@@ -184,10 +179,12 @@ def get_ip_ranges() -> tuple[list[str], list[str]]:
         _uri = 'https://api.cloudflare.com/client/v4/ips'
         _headers = {'Accept': 'application/json', 'X-Auth-Email': EMAIL}
         try:
-            response = await send_request(api_url=_uri, headers=_headers)
-            ipv4_cidrs = response.get('result', {}).get('ipv4_cidrs', DEFAULT_IPV4)
-            ipv6_cidrs = response.get('result', {}).get('ipv6_cidrs', [])
-            return ipv4_cidrs, ipv6_cidrs
+            with requests.Session() as send_request:
+                response = send_request.get(url=_uri, headers=_headers)
+                response_data: dict[str, dict[str, str] | list[str]] = response.json()
+                ipv4_cidrs = response_data.get('result', {}).get('ipv4_cidrs', DEFAULT_IPV4)
+                ipv6_cidrs = response_data.get('result', {}).get('ipv6_cidrs', [])
+                return ipv4_cidrs, ipv6_cidrs
 
         except CloudFlareAPIError:
             return [], []

src/firewall/__init__.py

@@ -23,7 +23,7 @@ def create_blog_url():
     server_url = config_instance().SERVER_NAME
     scheme = "http://" if "local" in server_url else "https://"
     blog_url = f"{scheme}{server_url}/blog/"
-    main_logger.info("Blog URL: {}".format(blog_url))
+    main_logger.info("Blog URL: {}".format(blog_url[:-1]))
     return blog_url
 
 

src/main.py

@@ -458,6 +458,14 @@ <h3 class="card-title"> Quora Community</h3>
                  <li class="font-weight-bold text-info"><a href="https://eodstockmarketapi.quora.com/" target="_blank">Join our Quora Community</a></li>
             </ul>
         </div>
+        <div class="card-body">
+            <div class="card-header">
+                <h3 class="card-title"> Slack Community </h3>
+                <ul class="card-footer">
+                    <li class="font-weight-bold"><a href="https://join.slack.com/t/eod-stock-apisite/shared_invite/zt-1uelcf229-c_6QAgWFNyVfXKZr1hYYoQ" target="_blank"> Join our Slack Community</a></li>
+                </ul>
+            </div>
+        </div>
 
       </div>
     </div>