fix: Solve problem with Maven packages due to ':' in SMT vars

Author: GermanMT

app/apis/managers/cargo_service.py

@@ -10,7 +10,7 @@
 from app.utils.others import looks_like_repo, normalize_repo_url, order_versions
 
 
-async def get_cargo_url_vendor(response: dict[str, Any]) -> tuple[str, str]:     
+async def get_cargo_url_vendor(response: dict[str, Any]) -> tuple[str, str]:
     raw_url = response.get("crate", {}).get("repository")
     norm_url = await normalize_repo_url(raw_url)
     if norm_url and await looks_like_repo(norm_url):

app/apis/managers/maven_service.py

@@ -1,4 +1,5 @@
 from asyncio import TimeoutError, sleep
+from datetime import datetime
 from typing import Any
 from xml.etree.ElementTree import ParseError, fromstring
 
@@ -8,7 +9,6 @@
 from app.http_session import get_session
 from app.logger import logger
 from app.utils.others import looks_like_repo, normalize_repo_url, order_versions
-from datetime import datetime
 
 
 async def get_maven_url_vendor(group_id: str, artifact_id: str, version: str) -> tuple[str, str]:
@@ -158,8 +158,8 @@ async def get_maven_package(group_id: str, artifact_id: str, version_name: str)
                 dep_group_id = dep.find("mvn:groupId", namespace).text
                 dep_artifact_id = dep.find("mvn:artifactId", namespace).text
                 dep_version = dep.find("mvn:version", namespace)
-                dep_version_text = dep_version.text if dep_version is not None else "latest"
-                if not any(char in dep_version_text for char in ["[", "]", "(", ")"]):
+                dep_version_text = dep_version.text if dep_version is not None else "any"
+                if dep_version_text != "any" and not any(char in dep_version_text for char in ["[", "]", "(", ")"]):
                     dep_version_text = "[" + dep_version_text + "]"
                 requirement[f"{dep_group_id}:{dep_artifact_id}"] = dep_version_text
         except ParseError:

app/apis/managers/nuget_service.py

@@ -55,7 +55,7 @@ async def get_nuget_versions(package_name: str) -> tuple[list[dict[str, Any]], l
                 raw_url = catalog_entry.get("repositoryUrl")
                 norm_url = await normalize_repo_url(raw_url)
                 if norm_url and await looks_like_repo(norm_url):
-                    repository_url = norm_url 
+                    repository_url = norm_url
                     vendor = norm_url.split("/")[-2]
                 raw_versions.append({
                     "name": name,

app/utils/others/normalize_repo_url.py

@@ -32,5 +32,5 @@ async def normalize_repo_url(url: str | None) -> str | None:
     if git_match:
         host = parsed.netloc.lower()
         if any(platform in host for platform in ['github.com', 'gitlab.com', 'bitbucket.org']):
-            clean = git_match.group(1)    
+            clean = git_match.group(1)
     return clean

app/utils/repo_analyzer/requirement_files/pom_xml_analyzer.py

@@ -29,7 +29,7 @@ async def analyze_pom_xml(
             if version_text.startswith("${") and version_text.endswith("}"):
                 property_key = version_text[2:-1]
                 version_text = properties.get(property_key, "any")
-            if not any(char in version_text for char in ["[", "]", "(", ")"]):
+            if version_text != "any" and not any(char in version_text for char in ["[", "]", "(", ")"]):
                 version_text = f"[{version_text}]"
             requirement_files[requirement_file_name]["requirement"][f"{group_id_text}:{artifact_id_text}"] = version_text
     except Exception:

app/utils/smt/model/smt_model.py

@@ -40,8 +40,8 @@ async def transform(self) -> str:
         str_sum = await self.sum()
         self.ctc_domain += f"(= {file_risk_name} {str_sum})"
         for indirect_var in self.indirect_vars:
-            self.var_domain.add(f"(declare-const {indirect_var} Int)")
-            self.var_domain.add(f"(declare-const impact_{indirect_var} Real)")
+            self.var_domain.add(f"(declare-const |{indirect_var}| Int)")
+            self.var_domain.add(f"(declare-const |impact_{indirect_var}| Real)")
         model_text = f"{' '.join(self.var_domain)} (assert (and {self.ctc_domain}))"
         self.domain = parse_smt2_string(model_text)
         self.func_obj = Real(file_risk_name)
@@ -52,10 +52,10 @@ async def transform_direct_package(self, require: dict[str, Any]) -> None:
         filtered_versions = await filter_versions(self.node_type, self.source_data["have"][require["package"]], require["constraints"])
         versions_impacts: dict[int, int] = {version.get("serial_number"): version[self.aggregator] for version in filtered_versions}
         versions_names = list(versions_impacts.keys())
-        self.directs.append(require["package"])
-        var_impact = f"impact_{require['package']}"
+        self.directs.append(f"|{require["package"]}|")
+        var_impact = f"|impact_{require['package']}|"
         self.impacts.add(var_impact)
-        self.var_domain.add(f"(declare-const {require['package']} Int)")
+        self.var_domain.add(f"(declare-const |{require['package']}| Int)")
         self.var_domain.add(f"(declare-const {var_impact} Real)")
         await self.build_direct_contraint(
             require["package"], versions_names
@@ -85,7 +85,7 @@ async def transform_versions(self, versions: dict[int, int], var: str, require:
         ):
             _default = {}
             if require:
-                self.impacts.add(f"impact_{require['package']}")
+                self.impacts.add(f"|impact_{require['package']}|")
                 self.indirect_vars.add(var)
                 self.indirect_vars.add(require["parent_version_name"])
                 _default = {.0: {-1}}
@@ -119,13 +119,13 @@ async def build_indirect_constraints(self) -> None:
                 self.ctc_domain += f"(=> {await self.group_versions(parent, list(parent_versions), True)} {versions}) "
         for child, _ in self.parents.items():
             for parent, parent_versions in _.items():
-                self.ctc_domain += f"(=> (not {await self.group_versions(parent, list(parent_versions), True)}) (= {child} -1)) "
+                self.ctc_domain += f"(=> (not {await self.group_versions(parent, list(parent_versions), True)}) (= |{child}| -1)) "
 
 
     async def build_impact_constraints(self) -> None:
         for var, _ in self.ctcs.items():
             for impact, versions in _.items():
-                self.ctc_domain += f"(=> {await self.group_versions(var, list(versions), True)} (= impact_{var} {impact})) "
+                self.ctc_domain += f"(=> {await self.group_versions(var, list(versions), True)} (= |impact_{var}| {impact})) "
 
 
     async def group_versions(
@@ -143,9 +143,9 @@ async def group_versions(
             if versions[i] == versions[i - 1] + step:
                 current_group.append(versions[i])
             else:
-                constraints.append(await self.create_constraint_for_group(var, current_group, ascending))
+                constraints.append(await self.create_constraint_for_group(f"|{var}|", current_group, ascending))
                 current_group = [versions[i]]
-        constraints.append(await self.create_constraint_for_group(var, current_group, ascending))
+        constraints.append(await self.create_constraint_for_group(f"|{var}|", current_group, ascending))
         return constraints[0] if len(constraints) == 1 else f"(or {' '.join(constraints)})"