Skip to content

Commit e0137a5

Browse files
alielaliel
committed
refactor checks
1 parent 11f3d27 commit e0137a5

File tree

1 file changed

+78
-73
lines changed

1 file changed

+78
-73
lines changed

‎src/aleph/sdk/domain.py‎

Lines changed: 78 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -35,94 +35,99 @@ async def get_dnslink(self, url: str):
3535
if query is not None and len(query) > 0:
3636
return query[0].text
3737

38-
async def get_control(self, url: str):
39-
domain = self.url_to_domain(url)
40-
query = await self.query(f"_control.{domain}", "TXT")
41-
if query is not None and len(query) > 0:
42-
return query[0].text
43-
4438
async def check_domain_configured(self, domain, target, owner):
4539
try:
4640
print("Check...", target)
4741
return await self.check_domain(domain, target, owner)
4842
except Exception as error:
4943
raise DomainConfigurationError(error)
5044

51-
async def check_domain(self, url: str, target: str, owner: Optional[str] = None):
52-
return await self.check_ipfs_domain(url, target, owner)
53-
54-
55-
async def check_ipfs_domain(
45+
async def check_domain(
5646
self, url: str, target: str, owner: Optional[str] = None
5747
):
58-
status = {"cname": True, "owner_proof": False}
48+
status = {"cname": False, "owner_proof": False}
5949

6050
target = target.lower()
6151
domain = self.url_to_domain(url)
6252

63-
if target == "ipfs":
64-
status["delegation"] = False
53+
dns_rules = self.get_required_dns_rules(url, target, owner)
6554

66-
# check1: CNAME value should be ipfs or program
67-
res = await self.query(domain, "CNAME")
68-
if target.lower() == "ipfs":
69-
expected_value = settings.DNS_IPFS_DOMAIN
70-
else:
71-
expected_value = settings.DNS_PROGRAM_DOMAIN
72-
73-
assert_error = (
74-
f"CNAME record not found: {domain}",
75-
f"Create a CNAME record for {domain} with values {expected_value}",
76-
status,
77-
)
78-
if res is None or not hasattr(res, "cname"):
79-
raise DomainConfigurationError(assert_error)
80-
81-
assert_error = (
82-
f"{domain} should have a valid CNAME value, {res.cname} provided",
83-
f"Create a CNAME record for {domain} with values {expected_value}",
84-
status,
85-
)
86-
if res.cname != expected_value:
87-
raise DomainConfigurationError(assert_error)
88-
89-
status["cname"] = True
90-
if target.lower() == "ipfs":
91-
# check2: CNAME value of _dnslink.__custom_domain__
92-
# should be _dnslink.__custom_domain__.static.public.aleph.sh
93-
res = await self.query(f"_dnslink.{domain}", "CNAME")
94-
95-
expected_value = f"_dnslink.{domain}.{settings.DNS_ROOT_DOMAIN}"
96-
assert_error = (
97-
f"CNAME record not found: _dnslink.{domain}",
98-
f"Create a CNAME record for _dnslink.{domain} with value: {expected_value}",
99-
status,
100-
)
101-
if res is None or not hasattr(res, "cname") or res.cname != expected_value:
102-
raise DomainConfigurationError(assert_error)
103-
104-
status["delegation"] = True
105-
106-
# check3: TXT value of _control.__custom_domain__ should be the address of the owner
107-
owner_address = await self.get_control(domain)
108-
assert_error = (
109-
f"TXT record not found: _control.{domain}",
110-
f'Create a TXT record for _control.{domain} with value = "owner address"',
111-
status,
112-
)
113-
if owner_address is None:
114-
raise DomainConfigurationError(assert_error)
115-
116-
if owner is not None:
117-
if owner == owner_address:
118-
status["owner_proof"] = True
119-
else:
120-
raise DomainConfigurationError(
121-
(
122-
f"Owner address mismatch, got: {owner} expected: {owner_address}",
123-
f"",
124-
status,
55+
for dns_rule in dns_rules:
56+
status[dns_rule["rule_name"]] = False
57+
58+
record_name = dns_rule["dns"]["name"]
59+
record_type = dns_rule["dns"]["type"]
60+
record_value = dns_rule["dns"]["value"]
61+
62+
res = await self.query(record_name, record_type.upper())
63+
64+
if record_type == "txt":
65+
found = False
66+
67+
for _res in res:
68+
if hasattr(_res, "text") and _res.text == record_value:
69+
found = True
70+
71+
if found == False:
72+
raise DomainConfigurationError(
73+
(dns_rule["info"], dns_rule["on_error"], status)
12574
)
75+
76+
elif res is None or not hasattr(res, record_type) or getattr(res, record_type) != record_value:
77+
raise DomainConfigurationError(
78+
(dns_rule["info"], dns_rule["on_error"], status)
12679
)
12780

81+
status[dns_rule["rule_name"]] = True
82+
12883
return status
84+
85+
def get_required_dns_rules(self, url, target, owner: Optional[str] = None):
86+
domain = self.url_to_domain(url)
87+
target = target.lower()
88+
dns_rules = []
89+
90+
if target == "ipfs":
91+
cname_value = settings.DNS_IPFS_DOMAIN
92+
else:
93+
cname_value = settings.DNS_PROGRAM_DOMAIN
94+
95+
# cname rule
96+
dns_rules.append({
97+
"rule_name": "cname",
98+
"dns": {
99+
"type": "cname",
100+
"name": domain,
101+
"value": cname_value
102+
},
103+
"info": f"Create a CNAME record for {domain} with value {cname_value}",
104+
"on_error": f"CNAME record not found: {domain}"
105+
})
106+
107+
if target == "ipfs":
108+
# ipfs rule
109+
dns_rules.append({
110+
"rule_name": "delegation",
111+
"dns": {
112+
"type": "cname",
113+
"name": f"_dnslink.{domain}",
114+
"value": f"_dnslink.{domain}.{settings.DNS_ROOT_DOMAIN}"
115+
},
116+
"info": f"Create a CNAME record for _dnslink.{domain} with value _dnslink.{domain}.{settings.DNS_ROOT_DOMAIN}",
117+
"on_error": f"CNAME record not found: _dnslink.{domain}"
118+
})
119+
120+
if owner:
121+
# ownership rule
122+
dns_rules.append({
123+
"rule_name": "owner_proof",
124+
"dns": {
125+
"type": "txt",
126+
"name": f"_control.{domain}",
127+
"value": owner
128+
},
129+
"info": f"Create a TXT record for _control.{domain} with value = owner address",
130+
"on_error": f"Owner address mismatch"
131+
})
132+
133+
return dns_rules

0 commit comments

Comments
 (0)