Merge pull request #14 from blacklanternsecurity/master-backup

3.0 Update

.github/dependabot.yml

@@ -0,0 +1,8 @@
+version: 2
+updates:
+  - package-ecosystem: "pip"
+    directory: "/"
+    schedule:
+      interval: "weekly"
+    target-branch: "master"
+    open-pull-requests-limit: 10

.github/workflows/tests.yml

@@ -36,13 +36,18 @@ jobs:
         run: |
           pip install poetry
           poetry install
+      - name: Install ASN database
+        run: |
+          pip install pyasn
+          pyasn_util_download.py --latestv46
+          pyasn_util_convert.py --single rib.*.bz2 asn.db
       - name: Run tests
         run: |
           poetry run pytest --exitfirst --disable-warnings --log-cli-level=DEBUG
   update:
     runs-on: ubuntu-latest
     needs: test
-    if: github.event_name == 'push' || github.event_name == 'schedule'
+    if: github.ref == 'refs/heads/master' && github.actor != 'dependabot[bot]' && (github.event_name == 'push' || github.event_name == 'schedule')
     steps:
       - uses: actions/checkout@v3
       - name: Set up Python
@@ -53,6 +58,11 @@ jobs:
         run: |
           pip install poetry
           poetry install
+      - name: Install ASN database
+        run: |
+          pip install pyasn
+          pyasn_util_download.py --latestv46
+          pyasn_util_convert.py --single rib.*.bz2 asn.db
       - name: Update cloud_providers.json
         run: poetry run cloudcheck forceupdate
       - name: git add cloud_providers.json
@@ -70,7 +80,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Dynamic version
-        run: "sed -i s/2.0.0.0/2.2.0.$(git rev-list HEAD --count)/g pyproject.toml"
+        run: "sed -i s/3.0.0.0/3.0.0.$(git rev-list HEAD --count)/g pyproject.toml"
       - uses: actions/setup-python@v4
         with:
           python-version: "3.x"

cloudcheck/cidr.py

@@ -9,8 +9,12 @@ class CidrRanges:
         https://github.com/yl2chen/cidranger
     """
 
-    def __init__(self, ranges):
+    def __init__(self, ranges=None):
         self.cidrs = set()
+        if ranges is not None:
+            self.update(ranges)
+
+    def update(self, ranges):
         for r in ranges:
             self.cidrs.add(ipaddress.ip_network(r, strict=False))
 

cloudcheck/providers/base.py

@@ -1,5 +1,6 @@
 import re
 import httpx
+import pyasn
 import logging
 import ipaddress
 import traceback
@@ -13,6 +14,12 @@
 log = logging.getLogger("cloudcheck.providers")
 
 
+try:
+    asndb = pyasn.pyasn("asn.db")
+except Exception:
+    asndb = None
+
+
 class CloudProviderJSON(BaseModel):
     name: str = ""
     domains: List[str] = []
@@ -22,6 +29,7 @@ class CloudProviderJSON(BaseModel):
     regexes: Dict[str, List[str]] = {}
     provider_type: str = "cloud"
     ips_url: str = ""
+    asns: List[int] = []
 
     @field_validator("cidrs")
     @classmethod
@@ -38,16 +46,18 @@ class BaseCloudProvider:
     headers = {
         "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/104.0.5112.79 Safari/537.36"
     }
+    asns = []
 
     def __init__(self, j, httpx_client=None):
         self._httpx_client = httpx_client
         self._log = None
+        self.ranges = CidrRanges()
         if j is not None:
             p = CloudProviderJSON(**j)
             self.domains = set(
                 [d.lower() for d in set(list(self.domains) + list(p.domains))]
             )
-            self.ranges = CidrRanges(p.cidrs)
+            self.ranges.update(p.cidrs)
             self.last_updated = p.last_updated
 
         self._bucket_name_regex = re.compile("^" + self.bucket_name_regex + "$", re.I)
@@ -63,17 +73,27 @@ def __init__(self, j, httpx_client=None):
 
     async def update(self):
         try:
-            response = await self.httpx_client.get(
-                self.ips_url, follow_redirects=True, headers=self.headers
-            )
-            ranges = self.parse_response(response)
-            if ranges:
-                self.ranges = CidrRanges(ranges)
-                self.last_updated = datetime.now()
+            self.last_updated = datetime.now()
+            self.ranges = CidrRanges(self.get_subnets())
+            if self.ips_url:
+                response = await self.httpx_client.get(
+                    self.ips_url, follow_redirects=True, headers=self.headers
+                )
+                ranges = self.parse_response(response)
+                if ranges:
+                    self.ranges.update(ranges)
         except Exception as e:
             log.warning(f"Error retrieving {self.ips_url}: {e}")
             log.warning(traceback.format_exc())
 
+    def get_subnets(self):
+        subnets = set()
+        if asndb is not None:
+            for asn in self.asns:
+                for subnet in asndb.get_as_prefixes(asn):
+                    subnets.add(ipaddress.ip_network(subnet, strict=False))
+        return subnets
+
     def to_json(self):
         """
         domains: List[str] = []
@@ -92,6 +112,7 @@ def to_json(self):
             regexes=self.regexes,
             provider_type=self.provider_type,
             ips_url=self.ips_url,
+            asns=self.asns,
             bucket_name_regex=self.bucket_name_regex,
         ).dict()
 

cloudcheck/providers/zoho.py

@@ -0,0 +1,23 @@
+from .base import BaseCloudProvider
+
+
+class Zoho(BaseCloudProvider):
+    domains = [
+        "zoho.com",
+        "zoho.com.au",
+        "zoho.eu",
+        "zoho.in",
+        "zohocdn.com",
+        "zohomeetups.com",
+        "zohomerchandise.com",
+        "zohopublic.com",
+        "zohoschools.com",
+        "zohostatic.com",
+        "zohostatic.in",
+        "zohouniversity.com",
+        "zohowebstatic.com",
+    ]
+    asns = [
+        2639,
+    ]
+    provider_type = "cloud"

cloudcheck/test/test_cloudcheck.py

@@ -58,6 +58,9 @@ async def test_cloudcheck():
 
     assert cloud_providers.last_updated
 
+    zoho = cloud_providers.providers["zoho"]
+    assert zoho.asns
+
 
 if __name__ == "__main__":
     asyncio.run(test_cloudcheck())