2.0.7 (2025-09-14)

------------------
**Changed:** f-strings to more robustly helpers.url_join()

Author: vladimirs-git

CHANGELOG.rst

@@ -13,6 +13,14 @@ Unreleased
 **New:** `FortiGateAPI.monitor` connectors, to work with all `Monitor API` endpoints.
 
 
+2.0.7 (2025-09-14)
+------------------
+
+**Changed:** f-strings to more robustly helpers.url_join()
+
+**Added:** _init_host() Init host: valid hostname or valid IPv4/IPv6 address.
+
+
 2.0.6 (2025-05-17)
 ------------------
 

docs/conf.py

@@ -8,7 +8,7 @@
 project = "fortigate-api"
 copyright = "2021, Vladimirs Prusakovs"
 author = "Vladimirs Prusakovs"
-release = "2.0.6"
+release = "2.0.7"
 
 extensions = [
     "sphinx.ext.autodoc",

fortigate_api/cmdb/firewall/policy.py

@@ -58,6 +58,6 @@ def move(self, policyid: StrInt, position: str, neighbor: StrInt) -> Response:
             # "secretkey": self.fortigate.password,
             position: neighbor,
         }
-        url = f"{self.url}/{policyid}"
+        url = h.url_join(self.url, policyid)
         url = h.join_url_params(url, **params)
         return self.fortigate.put(url=url, data={})

fortigate_api/connector.py

@@ -31,7 +31,7 @@ def __init__(self, fortigate: FortiGate, **kwargs):
     @property
     def url(self) -> str:
         """URL to the fortigate-object."""
-        return f"{self.fortigate.url}/{self._path}"
+        return h.url_join(self.fortigate.url, self._path)
 
     def create(self, data: DAny) -> Response:
         """Create the fortigate-object in the Fortigate.
@@ -79,7 +79,7 @@ def delete(
         if filter:
             return self._delete_by_filter(filter)
         uid = h.quote(uid)
-        url = f"{self.url}/{uid}"
+        url = h.url_join(self.url, uid)
         return self.fortigate.delete(url=url)
 
     def get(self, **kwargs) -> LDAny:
@@ -93,7 +93,7 @@ def get(self, **kwargs) -> LDAny:
         :rtype: List[dict]
         """
         uid = h.quote(vdict.pop(kwargs, key=self.uid))
-        url = f"{self.url}/{uid}".rstrip("/")
+        url = h.url_join(self.url, uid)
         url = h.join_url_params(url=url, **kwargs)
         if self.uid:
             items: LDAny = self.fortigate.get_results(url)
@@ -114,7 +114,7 @@ def is_exist(self, uid: StrInt) -> bool:
         uid = h.quote(uid)
         if not uid:
             raise ValueError("uid is required.")
-        url = f"{self.url}/{uid}"
+        url = h.url_join(self.url, uid)
         response = self.fortigate.exist(url)
         return response.ok
 
@@ -131,7 +131,7 @@ def update(self, data: DAny) -> Response:
         :rtype: Response
         """
         uid: str = self._get_uid(data)
-        url = f"{self.url}/{uid}".rstrip("/")
+        url = h.url_join(self.url, uid)
         return self.fortigate.put(url=url, data=data)
 
     # noinspection PyShadowingBuiltins
@@ -153,7 +153,7 @@ def _delete_by_filter(self, filter: UStr) -> Response:  # pylint: disable=redefi
         items: LDAny = self.get(filter=filters)
         for data in items:
             uid = h.quote(data[self.uid])
-            url = f"{self.url}/{uid}"
+            url = h.url_join(self.url, uid)
             response = self.fortigate.delete(url=url)
             responses.append(response)
         return h.highest_response(responses)
@@ -168,7 +168,7 @@ def _get_uid(self, data) -> str:
         if not self.uid:
             return ""
         if self.uid not in data:
-            raise ValueError(f"{self.uid} value is required.")
+            raise ValueError(f"uid={self.uid!r} value is required in data.")
         uid = str(data[self.uid])
         uid = h.quote(uid)
         return uid

fortigate_api/extended_filters.py

@@ -5,6 +5,7 @@
 from functools import wraps
 from ipaddress import ip_network, IPv4Network
 from typing import Tuple
+from urllib.parse import urljoin
 
 from vhelpers import vre
 
@@ -62,8 +63,10 @@ def efilter_by_sdst(policies: LDAny, efilter: str, connector) -> None:
         return
     # get addresses and address-groups from the Fortigate
     fortigate = connector.fortigate
-    addresses: LDAny = fortigate.get_results(url=f"{fortigate.url}/api/v2/cmdb/firewall/address")
-    addr_groups: LDAny = fortigate.get_results(url=f"{fortigate.url}/api/v2/cmdb/firewall/addrgrp")
+    url = urljoin(fortigate.url, "/api/v2/cmdb/firewall/address")
+    addresses: LDAny = fortigate.get_results(url)
+    url = urljoin(fortigate.url, "/api/v2/cmdb/firewall/addrgrp")
+    addr_groups: LDAny = fortigate.get_results(url)
     names_subnets_d: DLStr = _get_names_subnets(addresses, addr_groups)
     names_ipnets_d: DLInet = _convert_subnets_to_ipnets(names_subnets_d)
 

fortigate_api/fortigate_base.py

@@ -2,11 +2,11 @@
 
 from __future__ import annotations
 
+import ipaddress
 import json
 import logging
-import re
 from typing import Callable, Iterable, Optional
-from urllib.parse import urlencode, urljoin
+from urllib.parse import urlencode, urljoin, urlunparse
 
 import requests
 from requests import Session, Response
@@ -19,6 +19,7 @@
 # noinspection PyUnresolvedReferences
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 
+HTTP = "http"
 HTTPS = "https"
 PORT_443 = 443
 PORT_80 = 80
@@ -59,7 +60,7 @@ def __init__(self, **kwargs):
         :param bool logging_error: Logging only the REST API response with error.
             `True` - Enable errors logging, `False` - otherwise. Default is `False`.
         """
-        self.host = str(kwargs.get("host"))
+        self.host = _init_host(**kwargs)
         self.username = str(kwargs.get("username"))
         self.password = str(kwargs.get("password"))
         self.token = _init_token(**kwargs)
@@ -81,7 +82,7 @@ def __repr__(self):
         host = self.host
         username = self.username
         scheme = self.scheme
-        port = self.port if not (scheme == HTTPS and self.port == PORT_443) else ""
+        port = self.port
         timeout = self.timeout
         verify = self.verify
         vdom = self.vdom
@@ -123,11 +124,8 @@ def is_connected(self) -> bool:
     @property
     def url(self) -> str:
         """Return URL to the Fortigate."""
-        if self.scheme == HTTPS and self.port == 443:
-            return f"{self.scheme}://{self.host}"
-        if self.scheme == "http" and self.port == 80:
-            return f"{self.scheme}://{self.host}"
-        return f"{self.scheme}://{self.host}:{self.port}"
+        components = (self.scheme, f"{self.host}:{self.port}", "/", "", "", "")
+        return urlunparse(components)
 
     # ============================ login =============================
 
@@ -145,7 +143,7 @@ def login(self) -> None:
         if self.token:
             try:
                 response: Response = session.get(
-                    url=f"{self.url}/api/v2/monitor/system/status",
+                    url=urljoin(self.url, "/api/v2/monitor/system/status"),
                     headers=self._bearer_token(),
                     verify=self.verify,
                 )
@@ -158,7 +156,7 @@ def login(self) -> None:
         # password
         try:
             response = session.post(
-                url=f"{self.url}/logincheck",
+                url=urljoin(self.url, "/logincheck"),
                 data=urlencode([("username", self.username), ("secretkey", self.password)]),
                 timeout=self.timeout,
                 verify=self.verify,
@@ -182,13 +180,12 @@ def logout(self) -> None:
             if not self.token:
                 try:
                     self._session.get(
-                        url=f"{self.url}/logout",
+                        url=urljoin(self.url, "/logout"),
                         timeout=self.timeout,
                         verify=self.verify,
                     )
                 except SSLError:
                     pass
-            del self._session
         self._session = None
 
     # =========================== helpers ============================
@@ -251,7 +248,7 @@ def _init_port(self, **kwargs) -> int:
         """Init port, 443 for scheme=`https`, 80 for scheme=`http`."""
         if port := int(kwargs.get("port") or 0):
             return port
-        if self.scheme == "http":
+        if self.scheme == HTTP:
             return PORT_80
         return PORT_443
 
@@ -281,7 +278,7 @@ def _response(self, method: Method, url: str, data: ODAny = None) -> Response:
         :rtype: Response
         """
         params: DAny = {
-            "url": self._valid_url(url),
+            "url": urljoin(self.url, url),
             "params": urlencode([("vdom", self.vdom)]),
             "timeout": self.timeout,
             "verify": self.verify,
@@ -299,26 +296,37 @@ def _response(self, method: Method, url: str, data: ODAny = None) -> Response:
             raise self._hide_secret_ex(ex)
         return response
 
-    def _valid_url(self, url: str) -> str:
-        """Return a valid URL string.
 
-        Add `https://` to `url` if it is absent and remove trailing `/` character.
-        """
-        if re.match("http(s)?://", url):
-            return url.rstrip("/")
-        path = url.strip("/")
-        return urljoin(self.url, path)
+# =========================== helpers ============================
 
 
-# =========================== helpers ============================
+def _init_host(**kwargs) -> str:
+    """Init host: valid hostname or valid IPv4/IPv6 address."""
+    host = str(kwargs.get("host", "")).strip()
+    if not host:
+        raise ValueError(f"{host=!r}, hostname is not specified.")
+
+    # Strip brackets if provided (IPv6 URL style)
+    if host.startswith("[") and host.endswith("]"):
+        host = host[1:-1]
+
+    # Try IP validation first
+    try:
+        ip = ipaddress.ip_address(host)
+        if isinstance(ip, ipaddress.IPv6Address):
+            host = f"[{host}]"
+    except ValueError:
+        pass  # hostname
+
+    return host
 
 
 def _init_scheme(**kwargs) -> str:
     """Init scheme `https` or `http`."""
     scheme = str(kwargs.get("scheme") or HTTPS)
-    expected = ["https", "http"]
+    expected = [HTTPS, HTTP]
     if scheme not in expected:
-        raise ValueError(f"{scheme=}, {expected=}.")
+        raise ValueError(f"{scheme=!r}, {expected=!r}.")
     return scheme
 
 
@@ -328,7 +336,7 @@ def _init_token(**kwargs) -> str:
     if not token:
         return ""
     if kwargs.get("username"):
-        raise ValueError("Mutually excluded: username, token.")
+        raise ValueError("A username and a token are mutually exclusive.")
     if kwargs.get("password"):
-        raise ValueError("Mutually excluded: password, token.")
+        raise ValueError("A password and a token are mutually exclusive.")
     return token

fortigate_api/helpers.py

@@ -251,9 +251,9 @@ def join_url_params(url: str, **params) -> str:
         return: "https://fomain.com?a=a&b=b&b=B"
     """
     url_o: ParseResult = urlparse(url)
-    params_or: DAny = parse_qs(url_o.query)
-    params_: DAny = {**params_or, **params}
-    query: str = urlencode(params_, doseq=True)
+    params_old: DAny = parse_qs(url_o.query)
+    params_new: DAny = {**params_old, **params}
+    query: str = urlencode(params_new, doseq=True)
     url_o = url_o._replace(query=query)
     return url_o.geturl()
 
@@ -270,6 +270,14 @@ def quote(string: Any) -> str:
     return parse.quote(string=string, safe="")
 
 
+def url_join(base: str, path: StrInt) -> str:
+    """Join path segments to a base URL safely using slash."""
+    path_ = str(path)
+    if not path_:
+        return base
+    return base.rstrip("/") + "/" + path_.lstrip("/")
+
+
 def url_to_app_model(url: str) -> str:
     """Parse app/model name from the URL.
 

fortigate_api/schema/custom/cmdb/firewall/policy.py

@@ -58,6 +58,6 @@ def move(self, policyid: StrInt, position: str, neighbor: StrInt) -> Response:
             # "secretkey": self.fortigate.password,
             position: neighbor,
         }
-        url = f"{self.url}/{policyid}"
+        url = h.url_join(self.url, policyid)
         url = h.join_url_params(url, **params)
         return self.fortigate.put(url=url, data={})

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "fortigate_api"
-version = "2.0.6"
+version = "2.0.7"
 description = "Python package to configure Fortigate (Fortios) devices using REST API and SSH"
 authors = ["Vladimirs Prusakovs <[email protected]>"]
 readme = "README.rst"