Repository: apache/airflow
File: airflow-core/src/airflow/models/connection.py (Python)

Code before:
        except json.JSONDecodeError:
            raise ValueError(f"Encountered non-JSON in `extra` field for connection {conn_id!r}.")
        return None

    @reconstructor
    def on_db_load(self):
        if self.password:
            mask_secret(self.password)
            mask_secret(quote(self.password))

    @staticmethod
    def _normalize_conn_type(conn_type):
        if conn_type == "postgresql":
            conn_type = "postgres"
        elif "-" in conn_type:
            conn_type = conn_type.replace("-", "_")
        return conn_type

    def _parse_from_uri(self, uri: str):
        uri_match = RE_SAFE_LOG_URI.search(uri)
        if uri_match:
            # Create sanitised uri for logging
            pwd = uri_match.group(2)
            safe_log_uri = uri.replace(pwd, "******")
        else:  # Assume no password in URI
            safe_log_uri = uri
        schemes_count_in_uri = uri.count("://")
        if schemes_count_in_uri > 2:
            raise AirflowException(f"Invalid connection string: {safe_log_uri}.")
        host_with_protocol = schemes_count_in_uri == 2
        uri_parts = urlsplit(uri)
        conn_type = uri_parts.scheme
        self.conn_type = self._normalize_conn_type(conn_type)
        rest_of_the_url = uri.replace(f"{conn_type}://", ("" if host_with_protocol else "//"))
        if host_with_protocol:
            uri_splits = rest_of_the_url.split("://", 1)
            if "@" in uri_splits[0] or ":" in uri_splits[0]:
                raise AirflowException(f"Invalid connection string: {safe_log_uri}.")
        uri_parts = urlsplit(rest_of_the_url)
        protocol = uri_parts.scheme if host_with_protocol else None
        host = _parse_netloc_to_hostname(uri_parts)
        self.host = self._create_host(protocol, host)
        quoted_schema = uri_parts.path[1:]
        self.schema = unquote(quoted_schema) if quoted_schema else quoted_schema
        self.login = unquote(uri_parts.username) if uri_parts.username else uri_parts.username
        self.password = unquote(uri_parts.password) if uri_parts.password else uri_parts.password
        self.port = uri_parts.port
        if uri_parts.query:
            query = dict(parse_qsl(uri_parts.query, keep_blank_values=True))
            if self.EXTRA_KEY in query:
                self.extra = query[self.EXTRA_KEY]

Review comment (suggestion):
Instead of using regex to create your own parser why not use `urllib.parse` instead? You could do something like this:

```

def _parse_from_uri(self, uri: str):

    parsed = urlsplit(uri)
    
    if parsed.password is not None:
        username = parsed.username or ""
        hostname = parsed.hostname or ""
        port = f":{parsed.port}" if parsed.port else ""    
        masked_netloc = f"{username}:******@{hostname}{port}"
    
        safe_log_uri = urlunsplit(
            (
                parsed.scheme,
                masked_netloc,
                parsed.path,
                parsed.query,
                parsed.fragment,
            )
        )
    else:
        safe_log_uri = uri
  ```
Keep in mind that the above is just a suggested appraoch. It will need to be validated and tested.

Code after:
        return None

    @reconstructor
    def on_db_load(self):
        if self.password:
            mask_secret(self.password)
            mask_secret(quote(self.password))

    @staticmethod
    def _normalize_conn_type(conn_type):
        if conn_type == "postgresql":
            conn_type = "postgres"
        elif "-" in conn_type:
            conn_type = conn_type.replace("-", "_")
        return conn_type

    def _parse_from_uri(self, uri: str):
        schemes_count_in_uri = uri.count("://")
        if schemes_count_in_uri > 2:
            raise AirflowException("Invalid connection string.")
        host_with_protocol = schemes_count_in_uri == 2
        uri_parts = urlsplit(uri)
        conn_type = uri_parts.scheme
        self.conn_type = self._normalize_conn_type(conn_type)
        rest_of_the_url = uri.replace(f"{conn_type}://", ("" if host_with_protocol else "//"))
        if host_with_protocol:
            uri_splits = rest_of_the_url.split("://", 1)
            if "@" in uri_splits[0] or ":" in uri_splits[0]:
                raise AirflowException("Invalid connection string.")
        uri_parts = urlsplit(rest_of_the_url)
        protocol = uri_parts.scheme if host_with_protocol else None
        host = _parse_netloc_to_hostname(uri_parts)
        self.host = self._create_host(protocol, host)
        quoted_schema = uri_parts.path[1:]
        self.schema = unquote(quoted_schema) if quoted_schema else quoted_schema
        self.login = unquote(uri_parts.username) if uri_parts.username else uri_parts.username
        self.password = unquote(uri_parts.password) if uri_parts.password else uri_parts.password
        self.port = uri_parts.port
        if uri_parts.query:
            query = dict(parse_qsl(uri_parts.query, keep_blank_values=True))
            if self.EXTRA_KEY in query:
                self.extra = query[self.EXTRA_KEY]
            else:
                for key, value in query.items():
                    try:
                        query[key] = json.loads(value)
                    except (JSONDecodeError, TypeError):
                        self.log.info("Failed parsing the json for key %s", key)
                self.extra = json.dumps(query)

    @staticmethod