feat: validate srv owner

api/app/controllers/api/record.py

@@ -136,7 +136,10 @@ def post(self):
             # rtype no need to be validated & no need to check its length
             # `get_typeid` will raise error for non existing rtype
             validator.validate(rtype, rdata)
-            validator.validate("owner", owner)
+            if rtype == "srv":
+                validator.validate("owner-srv", owner)
+            else:
+                validator.validate("owner", owner)
         except Exception as e:
             return response(422, message=f"{e}")
 

api/app/helpers/validator.py

@@ -159,6 +159,33 @@ def check_hypen(label):
         raise ValueError("Bad OWNER")
 
 
+def is_valid_srv_owner(owner):
+    """Check if it's a valid srv owner.
+
+    Rules:
+    - must contain service and protocol name
+    - can't contain the zone name
+    - enfore the service name to be prefixed with _
+    - enfore the protocol to be prefixed with _
+    """
+
+    if "." not in owner:
+        raise ValueError("Bad OWNER")
+
+    labels = owner.split(".")
+    if len(labels) < 2:
+        raise ValueError("Bad OWNER")
+
+    service_name = labels[0]
+    protocol_type = labels[1]
+    if not service_name.startswith("_"):
+        raise ValueError("Service name must start with underscore")
+    if not protocol_type.startswith("_"):
+        raise ValueError("Protocol type must start with underscore")
+
+    is_valid_owner(owner)
+
+
 functions = {
     "A": is_valid_ip,
     "AAAA": is_valid_ip,
@@ -171,6 +198,7 @@ def check_hypen(label):
     "TXT": is_valid_txt,
     "SRV": is_valid_srv,
     "OWNER": is_valid_owner,
+    "OWNER-SRV": is_valid_srv_owner,
 }
 
 

api/tests/unit/test_validator.py

@@ -135,6 +135,20 @@ def test_valid_owner():
         validator.is_valid_owner("a" * 256)
 
 
+def test_valid_srv_owner():
+    validator.is_valid_srv_owner("_xmpp._tcp")
+
+    with pytest.raises(Exception):
+        # owner ends with dot
+        validator.is_valid_srv_owner("_xmpp._tcp.example.com.")
+    with pytest.raises(Exception):
+        # service name doesn't prefixed with _
+        validator.is_valid_srv_owner("xmpp._tcp")
+    with pytest.raises(Exception):
+        # protocol type doesn't prefixed with _
+        validator.is_valid_srv_owner("_xmpp.tcp")
+
+
 def test_validate_func():
     # validator exists
     validator.validate("A", "192.0.2.1")