refactor

Author: aliel

src/aleph/sdk/conf.py

@@ -34,10 +34,10 @@ class Settings(BaseSettings):
     CODE_USES_SQUASHFS: bool = which("mksquashfs") is not None  # True if command exists
 
     # Dns resolver
-    IPFS_DOMAINS = "ipfs.public.aleph.sh"
-    PROGRAM_DOMAINS = "program.public.aleph.sh"
-    ROOT_DOMAIN = "static.public.aleph.sh"
-    RESOLVERS = ["1.1.1.1", "8.8.8.8"]
+    DNS_IPFS_DOMAIN = "ipfs.public.aleph.sh"
+    DNS_PROGRAM_DOMAIN = "program.public.aleph.sh"
+    DNS_ROOT_DOMAIN = "static.public.aleph.sh"
+    DNS_RESOLVERS = ["1.1.1.1", "1.0.0.1"]
 
     class Config:
         env_prefix = "ALEPH_"

src/aleph/sdk/domain.py

@@ -7,7 +7,7 @@
 
 class AlephDNS:
     def __init__(self):
-        self.resolver = aiodns.DNSResolver(servers=settings.RESOLVERS)
+        self.resolver = aiodns.DNSResolver(servers=settings.DNS_RESOLVERS)
         self.fqdn_matcher = re.compile(r"https?://?")
 
     async def query(self, name: str, query_type: str):
@@ -41,69 +41,66 @@ async def get_control(self, url: str):
         if query is not None and len(query) > 0:
             return query[0].text
 
-    async def check_domain_configured(self, domain, _type, owner):
+    async def check_domain_configured(self, domain, target, owner):
         try:
-            print("Check...", _type)
-            return await self.check_domain(domain, _type, owner)
+            print("Check...", target)
+            return await self.check_domain(domain, target, owner)
         except Exception as error:
             raise DomainConfigurationError(error)
 
-    async def check_domain(self, url: str, _type: str, owner: Optional[str] = None):
-        # if _type.lower() == 'ipfs':
-        return await self.check_ipfs_domain(url, _type, owner)
-        # elif _type.lower() == 'program':
-        #    pass
+    async def check_domain(self, url: str, target: str, owner: Optional[str] = None):
+        return await self.check_ipfs_domain(url, target, owner)
+
 
     async def check_ipfs_domain(
-        self, url: str, _type: str, owner: Optional[str] = None
+        self, url: str, target: str, owner: Optional[str] = None
     ):
         status = {"cname": True, "owner_proof": False}
 
-        _type = _type.lower()
+        target = target.lower()
         domain = self.url_to_domain(url)
 
-        if _type == "ipfs":
+        if target == "ipfs":
             status["delegation"] = False
 
         # check1: CNAME value should be ipfs or program
         res = await self.query(domain, "CNAME")
-        if _type.lower() == "ipfs":
-            expected_value = settings.IPFS_DOMAINS
+        if target.lower() == "ipfs":
+            expected_value = settings.DNS_IPFS_DOMAIN
         else:
-            expected_value = settings.PROGRAM_DOMAINS
+            expected_value = settings.DNS_PROGRAM_DOMAIN
 
         assert_error = (
             f"CNAME record not found: {domain}",
             f"Create a CNAME record for {domain} with values {expected_value}",
             status,
         )
-
-        assert res is not None, assert_error
-        assert hasattr(res, "cname"), assert_error
+        if res is None or not hasattr(res, "cname"):
+            raise DomainConfigurationError(assert_error)
 
         assert_error = (
             f"{domain} should have a valid CNAME value, {res.cname} provided",
             f"Create a CNAME record for {domain} with values {expected_value}",
             status,
         )
-        assert res.cname in expected_value, assert_error
-        status["cname"] = True
+        if res.cname != expected_value:
+            raise DomainConfigurationError(assert_error)
 
-        if _type.lower() == "ipfs":
+        status["cname"] = True
+        if target.lower() == "ipfs":
             # check2: CNAME value of _dnslink.__custom_domain__
             # should be _dnslink.__custom_domain__.static.public.aleph.sh
             res = await self.query(f"_dnslink.{domain}", "CNAME")
 
-            expected_value = f"_dnslink.{domain}.{settings.ROOT_DOMAIN}"
+            expected_value = f"_dnslink.{domain}.{settings.DNS_ROOT_DOMAIN}"
             assert_error = (
                 f"CNAME record not found: _dnslink.{domain}",
                 f"Create a CNAME record for _dnslink.{domain} with value: {expected_value}",
                 status,
             )
+            if res is None or not hasattr(res, "cname") or res.cname != expected_value:
+                raise DomainConfigurationError(assert_error)
 
-            assert res is not None, assert_error
-            assert hasattr(res, "cname"), assert_error
-            assert res.cname == expected_value, assert_error
             status["delegation"] = True
 
         # check3: TXT value of _control.__custom_domain__ should be the address of the owner
@@ -113,14 +110,19 @@ async def check_ipfs_domain(
             f'Create a TXT record for _control.{domain} with value = "owner address"',
             status,
         )
-        assert owner_address is not None, assert_error
+        if owner_address is None:
+            raise DomainConfigurationError(assert_error)
 
         if owner is not None:
-            assert owner_address == owner, (
-                f"Owner address mismatch, got: {owner} expected: {owner_address}",
-                f"",
-                status,
-            )
-            status["owner_proof"] = True
+            if owner == owner_address:
+                status["owner_proof"] = True
+            else:
+                raise DomainConfigurationError(
+                    (
+                        f"Owner address mismatch, got: {owner} expected: {owner_address}",
+                        f"",
+                        status,
+                    )
+                )
 
         return status

tests/unit/test_domains.py

@@ -2,6 +2,7 @@
 import asyncio
 
 from aleph.sdk.domain import AlephDNS
+from aleph.sdk.exceptions import DomainConfigurationError
 
 
 @pytest.mark.asyncio
@@ -15,7 +16,7 @@ async def test_url_to_domain():
 
 
 @pytest.mark.asyncio
-async def test_get_ipv6_adress():
+async def test_get_ipv6_address():
     alephdns = AlephDNS()
     url = "https://aleph.im"
     ipv6_address = await alephdns.get_ipv6_address(url)
@@ -32,11 +33,17 @@ async def test_dnslink():
     assert dnslink is not None
 
 
-"""
 @pytest.mark.asyncio
-async def test_cname():
+async def test_configured_domain():
     alephdns = AlephDNS()
-    url = 'https://custom_domain_test.aleph.sh'
-    check = await alephdns.custom_domain_check(url)
-    assert check is not None
-"""
+    url = 'https://custom-domain-unit-test.aleph.sh'
+    status = await alephdns.check_domain(url, "ipfs", "0xfakeaddress")
+    assert type(status) is dict
+
+@pytest.mark.asyncio
+async def test_not_configured_domain():
+    alephdns = AlephDNS()
+    url = 'https://not-configured-domain.aleph.sh'
+    with pytest.raises(DomainConfigurationError):
+        status = await alephdns.check_domain(url, "ipfs", "0xfakeaddress")
+