Simplify pydantic model validation

src/middlewared/middlewared/api/base/types/__init__.py

@@ -2,4 +2,5 @@
 from .filesystem import *  # noqa
 from .iscsi import *  # noqa
 from .string import *  # noqa
+from .urls import *  # noqa
 from .user import *  # noqa

src/middlewared/middlewared/api/base/types/urls.py

@@ -0,0 +1,8 @@
+from typing import Annotated
+
+from pydantic import AfterValidator, HttpUrl
+
+from middlewared.api.base.validators import https_only_check
+
+
+HttpsOnlyURL = Annotated[HttpUrl, AfterValidator(https_only_check)]

src/middlewared/middlewared/api/base/validators.py

@@ -1,6 +1,8 @@
 from datetime import time
 import re
 
+from pydantic import HttpUrl
+
 
 def match_validator(pattern: re.Pattern, explanation: str | None = None):
     def validator(value: str):
@@ -21,3 +23,9 @@ def time_validator(value: str):
         except TypeError:
             raise ValueError('Time should be in 24 hour format like "18:00"')
     return value
+
+
+def https_only_check(url: HttpUrl) -> str:
+    if url.scheme != 'https':
+        raise ValueError('URL scheme must be https')
+    return str(url)

src/middlewared/middlewared/api/v25_04_0/tn_connect.py

@@ -1,7 +1,7 @@
-from pydantic import IPvAnyAddress, model_validator
+from pydantic import IPvAnyAddress
 
 from middlewared.api.base import BaseModel, ForUpdateMetaclass, NonEmptyString, single_argument_args
-from middlewared.utils.lang import undefined
+from middlewared.api.base.types import HttpsOnlyURL
 
 
 __all__ = [
@@ -18,28 +18,18 @@ class TNCEntry(BaseModel):
     status: NonEmptyString
     status_reason: NonEmptyString
     certificate: int | None
-    account_service_base_url: NonEmptyString
-    leca_service_base_url: NonEmptyString
-    tnc_base_url: NonEmptyString
+    account_service_base_url: HttpsOnlyURL
+    leca_service_base_url: HttpsOnlyURL
+    tnc_base_url: HttpsOnlyURL
 
 
 @single_argument_args('tn_connect_update')
 class TNCUpdateArgs(BaseModel, metaclass=ForUpdateMetaclass):
     enabled: bool
     ips: list[IPvAnyAddress]
-    account_service_base_url: NonEmptyString
-    leca_service_base_url: NonEmptyString
-    tnc_base_url: NonEmptyString
-
-    @model_validator(mode='after')
-    def validate_attrs(self):
-        for k in ('account_service_base_url', 'leca_service_base_url', 'tnc_base_url'):
-            value = getattr(self, k)
-            if value != undefined and not value.startswith('https://'):
-                raise ValueError(f'{k} must start with https://')
-            if value != undefined and not value.endswith('/'):
-                setattr(self, k, value + '/')
-        return self
+    account_service_base_url: HttpsOnlyURL
+    leca_service_base_url: HttpsOnlyURL
+    tnc_base_url: HttpsOnlyURL
 
 
 class TNCUpdateResult(BaseModel):

src/middlewared/middlewared/plugins/truenas_connect/update.py

@@ -101,7 +101,7 @@ async def do_update(self, data):
         db_payload = {
             'enabled': data['enabled'],
             'ips': data['ips'],
-        } | {k: data[k] for k in ('account_service_base_url', 'leca_service_base_url', 'tnc_base_url') if data.get(k)}
+        } | {k: data[k] for k in ('account_service_base_url', 'leca_service_base_url', 'tnc_base_url')}
         if config['enabled'] is False and data['enabled'] is True:
             # Finalization registration is triggered when claim token is generated
             # We make sure there is no pending claim token