565 lines
17 KiB
Python
565 lines
17 KiB
Python
# type: ignore
|
|
from __future__ import unicode_literals
|
|
|
|
from packaging.version import parse as parse_version
|
|
from packaging.specifiers import SpecifierSet
|
|
import requests
|
|
from typing import Any, Optional, Generator, Tuple, List
|
|
from safety.meta import get_meta_http_headers
|
|
|
|
from datetime import datetime
|
|
from dparse import parse, parser, updater, filetypes
|
|
from dparse.dependencies import Dependency
|
|
from dparse.parser import setuptools_parse_requirements_backport as parse_requirements
|
|
|
|
|
|
class RequirementFile(object):
|
|
"""
|
|
Class representing a requirements file with its content and metadata.
|
|
|
|
Attributes:
|
|
path (str): The file path.
|
|
content (str): The content of the file.
|
|
sha (Optional[str]): The SHA of the file.
|
|
"""
|
|
|
|
def __init__(self, path: str, content: str, sha: Optional[str] = None):
|
|
self.path = path
|
|
self.content = content
|
|
self.sha = sha
|
|
self._requirements: Optional[List] = None
|
|
self._other_files: Optional[List] = None
|
|
self._is_valid = None
|
|
self.is_pipfile = False
|
|
self.is_pipfile_lock = False
|
|
self.is_setup_cfg = False
|
|
|
|
def __str__(self) -> str:
|
|
return (
|
|
"RequirementFile(path='{path}', sha='{sha}', content='{content}')".format(
|
|
path=self.path,
|
|
content=self.content[:30] + "[truncated]"
|
|
if len(self.content) > 30
|
|
else self.content,
|
|
sha=self.sha,
|
|
)
|
|
)
|
|
|
|
@property
|
|
def is_valid(self) -> Optional[bool]:
|
|
"""
|
|
Checks if the requirements file is valid by parsing it.
|
|
|
|
Returns:
|
|
bool: True if the file is valid, False otherwise.
|
|
"""
|
|
if self._is_valid is None:
|
|
self._parse()
|
|
return self._is_valid
|
|
|
|
@property
|
|
def requirements(self) -> Optional[List]:
|
|
"""
|
|
Returns the list of requirements parsed from the file.
|
|
|
|
Returns:
|
|
List: The list of requirements.
|
|
"""
|
|
if not self._requirements:
|
|
self._parse()
|
|
return self._requirements
|
|
|
|
@property
|
|
def other_files(self) -> Optional[List]:
|
|
"""
|
|
Returns the list of other files resolved from the requirements file.
|
|
|
|
Returns:
|
|
List: The list of other files.
|
|
"""
|
|
if not self._other_files:
|
|
self._parse()
|
|
return self._other_files
|
|
|
|
@staticmethod
|
|
def parse_index_server(line: str) -> Optional[str]:
|
|
"""
|
|
Parses the index server from a given line.
|
|
|
|
Args:
|
|
line (str): The line to parse.
|
|
|
|
Returns:
|
|
str: The parsed index server.
|
|
"""
|
|
return parser.Parser.parse_index_server(line)
|
|
|
|
def _hash_parser(self, line: str) -> Optional[Tuple[str, List[str]]]:
|
|
"""
|
|
Parses the hashes from a given line.
|
|
|
|
Args:
|
|
line (str): The line to parse.
|
|
|
|
Returns:
|
|
List: The list of parsed hashes.
|
|
"""
|
|
return parser.Parser.parse_hashes(line)
|
|
|
|
def _parse_requirements_txt(self) -> None:
|
|
"""
|
|
Parses the requirements.txt file format.
|
|
"""
|
|
self.parse_dependencies(filetypes.requirements_txt)
|
|
|
|
def _parse_conda_yml(self) -> None:
|
|
"""
|
|
Parses the conda.yml file format.
|
|
"""
|
|
self.parse_dependencies(filetypes.conda_yml)
|
|
|
|
def _parse_tox_ini(self) -> None:
|
|
"""
|
|
Parses the tox.ini file format.
|
|
"""
|
|
self.parse_dependencies(filetypes.tox_ini)
|
|
|
|
def _parse_pipfile(self) -> None:
|
|
"""
|
|
Parses the Pipfile format.
|
|
"""
|
|
self.parse_dependencies(filetypes.pipfile)
|
|
self.is_pipfile = True
|
|
|
|
def _parse_pipfile_lock(self) -> None:
|
|
"""
|
|
Parses the Pipfile.lock format.
|
|
"""
|
|
self.parse_dependencies(filetypes.pipfile_lock)
|
|
self.is_pipfile_lock = True
|
|
|
|
def _parse_setup_cfg(self) -> None:
|
|
"""
|
|
Parses the setup.cfg format.
|
|
"""
|
|
self.parse_dependencies(filetypes.setup_cfg)
|
|
self.is_setup_cfg = True
|
|
|
|
def _parse(self) -> None:
|
|
"""
|
|
Parses the requirements file to extract dependencies and other files.
|
|
"""
|
|
self._requirements, self._other_files = [], []
|
|
if self.path.endswith(".yml") or self.path.endswith(".yaml"):
|
|
self._parse_conda_yml()
|
|
elif self.path.endswith(".ini"):
|
|
self._parse_tox_ini()
|
|
elif self.path.endswith("Pipfile"):
|
|
self._parse_pipfile()
|
|
elif self.path.endswith("Pipfile.lock"):
|
|
self._parse_pipfile_lock()
|
|
elif self.path.endswith("setup.cfg"):
|
|
self._parse_setup_cfg()
|
|
else:
|
|
self._parse_requirements_txt()
|
|
self._is_valid = len(self._requirements) > 0 or len(self._other_files) > 0
|
|
|
|
def parse_dependencies(self, file_type: str) -> None:
|
|
"""
|
|
Parses the dependencies from the content based on the file type.
|
|
|
|
Args:
|
|
file_type (str): The type of the file.
|
|
"""
|
|
result = parse(
|
|
self.content,
|
|
path=self.path,
|
|
sha=self.sha,
|
|
file_type=file_type,
|
|
marker=(
|
|
("pyup: ignore file", "pyup:ignore file"), # file marker
|
|
("pyup: ignore", "pyup:ignore"), # line marker
|
|
),
|
|
)
|
|
for dep in result.dependencies:
|
|
req = Requirement(
|
|
name=dep.name,
|
|
specs=dep.specs,
|
|
line=dep.line,
|
|
lineno=dep.line_numbers[0] if dep.line_numbers else 0,
|
|
extras=dep.extras,
|
|
file_type=file_type,
|
|
)
|
|
req.index_server = dep.index_server
|
|
if self.is_pipfile:
|
|
req.pipfile = self.path
|
|
req.hashes = dep.hashes
|
|
self._requirements.append(req)
|
|
self._other_files = result.resolved_files
|
|
|
|
def iter_lines(self, lineno: int = 0) -> Generator[str, None, None]:
|
|
"""
|
|
Iterates over lines in the content starting from a specific line number.
|
|
|
|
Args:
|
|
lineno (int): The line number to start from.
|
|
|
|
Yields:
|
|
str: The next line in the content.
|
|
"""
|
|
for line in self.content.splitlines()[lineno:]:
|
|
yield line
|
|
|
|
@classmethod
|
|
def resolve_file(cls, file_path: str, line: str) -> str:
|
|
"""
|
|
Resolves a file path from a given line.
|
|
|
|
Args:
|
|
file_path (str): The file path to resolve.
|
|
line (str): The line containing the file path.
|
|
|
|
Returns:
|
|
str: The resolved file path.
|
|
"""
|
|
return parser.Parser.resolve_file(file_path, line)
|
|
|
|
|
|
class Requirement(object):
|
|
"""
|
|
Class representing a single requirement.
|
|
|
|
Attributes:
|
|
name (str): The name of the requirement.
|
|
specs (SpecifierSet): The version specifiers for the requirement.
|
|
line (str): The line containing the requirement.
|
|
lineno (int): The line number of the requirement.
|
|
extras (List): The extras for the requirement.
|
|
file_type (str): The type of the file containing the requirement.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
specs: SpecifierSet,
|
|
line: str,
|
|
lineno: int,
|
|
extras: List,
|
|
file_type: str,
|
|
):
|
|
self.name = name
|
|
self.key = name.lower()
|
|
self.specs = specs
|
|
self.line = line
|
|
self.lineno = lineno
|
|
self.index_server = None
|
|
self.extras = extras
|
|
self.hashes = []
|
|
self.file_type = file_type
|
|
self.pipfile: Optional[str] = None
|
|
|
|
self.hashCmp = (
|
|
self.key,
|
|
self.specs,
|
|
frozenset(self.extras),
|
|
)
|
|
|
|
self._is_insecure = None
|
|
self._changelog = None
|
|
|
|
# Convert compatible releases to a range of versions
|
|
if (
|
|
len(self.specs._specs) == 1
|
|
and next(iter(self.specs._specs))._spec[0] == "~="
|
|
):
|
|
# convert compatible releases to something more easily consumed,
|
|
# e.g. '~=1.2.3' is equivalent to '>=1.2.3,<1.3.0', while '~=1.2'
|
|
# is equivalent to '>=1.2,<2.0'
|
|
min_version = next(iter(self.specs._specs))._spec[1]
|
|
max_version = list(parse_version(min_version).release)
|
|
max_version[-1] = 0
|
|
max_version[-2] = max_version[-2] + 1
|
|
max_version = ".".join(str(x) for x in max_version)
|
|
|
|
self.specs = SpecifierSet(">=%s,<%s" % (min_version, max_version))
|
|
|
|
def __eq__(self, other: Any) -> bool:
|
|
return isinstance(other, Requirement) and self.hashCmp == other.hashCmp
|
|
|
|
def __ne__(self, other: Any) -> bool:
|
|
return not self == other
|
|
|
|
def __str__(self) -> str:
|
|
return "Requirement.parse({line}, {lineno})".format(
|
|
line=self.line, lineno=self.lineno
|
|
)
|
|
|
|
def __repr__(self) -> str:
|
|
return self.__str__()
|
|
|
|
@property
|
|
def is_pinned(self) -> bool:
|
|
"""
|
|
Checks if the requirement is pinned to a specific version.
|
|
|
|
Returns:
|
|
bool: True if pinned, False otherwise.
|
|
"""
|
|
if (
|
|
len(self.specs._specs) == 1
|
|
and next(iter(self.specs._specs))._spec[0] == "=="
|
|
):
|
|
return True
|
|
return False
|
|
|
|
@property
|
|
def is_open_ranged(self) -> bool:
|
|
"""
|
|
Checks if the requirement has an open range of versions.
|
|
|
|
Returns:
|
|
bool: True if open ranged, False otherwise.
|
|
"""
|
|
if (
|
|
len(self.specs._specs) == 1
|
|
and next(iter(self.specs._specs))._spec[0] == ">="
|
|
):
|
|
return True
|
|
return False
|
|
|
|
@property
|
|
def is_ranged(self) -> bool:
|
|
"""
|
|
Checks if the requirement has a range of versions.
|
|
|
|
Returns:
|
|
bool: True if ranged, False otherwise.
|
|
"""
|
|
return len(self.specs._specs) >= 1 and not self.is_pinned
|
|
|
|
@property
|
|
def is_loose(self) -> bool:
|
|
"""
|
|
Checks if the requirement has no version specifiers.
|
|
|
|
Returns:
|
|
bool: True if loose, False otherwise.
|
|
"""
|
|
return len(self.specs._specs) == 0
|
|
|
|
@staticmethod
|
|
def convert_semver(version: str) -> dict:
|
|
"""
|
|
Converts a version string to a semantic version dictionary.
|
|
|
|
Args:
|
|
version (str): The version string.
|
|
|
|
Returns:
|
|
dict: The semantic version dictionary.
|
|
"""
|
|
semver = {"major": 0, "minor": 0, "patch": 0}
|
|
version_parts = version.split(".")
|
|
# don't be overly clever here. repitition makes it more readable and works exactly how
|
|
# it is supposed to
|
|
try:
|
|
semver["major"] = int(version_parts[0])
|
|
semver["minor"] = int(version_parts[1])
|
|
semver["patch"] = int(version_parts[2])
|
|
except (IndexError, ValueError):
|
|
pass
|
|
return semver
|
|
|
|
@property
|
|
def can_update_semver(self) -> bool:
|
|
"""
|
|
Checks if the requirement can be updated based on semantic versioning rules.
|
|
|
|
Returns:
|
|
bool: True if it can be updated, False otherwise.
|
|
"""
|
|
# return early if there's no update filter set
|
|
if "pyup: update" not in self.line:
|
|
return True
|
|
update = self.line.split("pyup: update")[1].strip().split("#")[0]
|
|
current_version = Requirement.convert_semver(
|
|
next(iter(self.specs._specs))._spec[1]
|
|
)
|
|
next_version = Requirement.convert_semver(self.latest_version) # type: ignore
|
|
if update == "major":
|
|
if current_version["major"] < next_version["major"]:
|
|
return True
|
|
elif update == "minor":
|
|
if (
|
|
current_version["major"] < next_version["major"]
|
|
or current_version["minor"] < next_version["minor"]
|
|
):
|
|
return True
|
|
return False
|
|
|
|
@property
|
|
def filter(self):
|
|
"""
|
|
Returns the filter for the requirement if specified.
|
|
|
|
Returns:
|
|
Optional[SpecifierSet]: The filter specifier set, or None if not specified.
|
|
"""
|
|
rqfilter = False
|
|
if "rq.filter:" in self.line:
|
|
rqfilter = self.line.split("rq.filter:")[1].strip().split("#")[0]
|
|
elif "pyup:" in self.line:
|
|
if "pyup: update" not in self.line:
|
|
rqfilter = self.line.split("pyup:")[1].strip().split("#")[0]
|
|
# unset the filter once the date set in 'until' is reached
|
|
if "until" in rqfilter:
|
|
rqfilter, until = [part.strip() for part in rqfilter.split("until")]
|
|
try:
|
|
until = datetime.strptime(until, "%Y-%m-%d")
|
|
if until < datetime.now():
|
|
rqfilter = False
|
|
except ValueError:
|
|
# wrong date formatting
|
|
pass
|
|
if rqfilter:
|
|
try:
|
|
(rqfilter,) = parse_requirements("filter " + rqfilter)
|
|
if len(rqfilter.specifier._specs) > 0:
|
|
return rqfilter.specifier
|
|
except ValueError:
|
|
pass
|
|
return False
|
|
|
|
@property
|
|
def version(self) -> Optional[str]:
|
|
"""
|
|
Returns the current version of the requirement.
|
|
|
|
Returns:
|
|
Optional[str]: The current version, or None if not pinned.
|
|
"""
|
|
if self.is_pinned:
|
|
return next(iter(self.specs._specs))._spec[1]
|
|
|
|
specs = self.specs
|
|
if self.filter:
|
|
specs = SpecifierSet(
|
|
",".join(
|
|
[
|
|
"".join(s._spec)
|
|
for s in list(specs._specs) + list(self.filter._specs)
|
|
]
|
|
)
|
|
)
|
|
return self.get_latest_version_within_specs( # type: ignore
|
|
specs,
|
|
versions=self.package.versions,
|
|
prereleases=self.prereleases, # type: ignore
|
|
)
|
|
|
|
def get_hashes(self, version: str) -> List:
|
|
"""
|
|
Retrieves the hashes for a specific version from PyPI.
|
|
|
|
Args:
|
|
version (str): The version to retrieve hashes for.
|
|
|
|
Returns:
|
|
List: A list of hashes for the specified version.
|
|
"""
|
|
headers = get_meta_http_headers()
|
|
r = requests.get(
|
|
"https://pypi.org/pypi/{name}/{version}/json".format(
|
|
name=self.key, version=version
|
|
),
|
|
headers=headers,
|
|
)
|
|
hashes = []
|
|
data = r.json()
|
|
|
|
for item in data.get("urls", {}):
|
|
sha256 = item.get("digests", {}).get("sha256", False)
|
|
if sha256:
|
|
hashes.append({"hash": sha256, "method": "sha256"})
|
|
return hashes
|
|
|
|
def update_version(
|
|
self, content: str, version: str, update_hashes: bool = True
|
|
) -> str:
|
|
"""
|
|
Updates the version of the requirement in the content.
|
|
|
|
Args:
|
|
content (str): The original content.
|
|
version (str): The new version to update to.
|
|
update_hashes (bool): Whether to update the hashes as well.
|
|
|
|
Returns:
|
|
str: The updated content.
|
|
"""
|
|
if self.file_type == filetypes.tox_ini:
|
|
updater_class = updater.ToxINIUpdater
|
|
elif self.file_type == filetypes.conda_yml:
|
|
updater_class = updater.CondaYMLUpdater
|
|
elif self.file_type == filetypes.requirements_txt:
|
|
updater_class = updater.RequirementsTXTUpdater
|
|
elif self.file_type == filetypes.pipfile:
|
|
updater_class = updater.PipfileUpdater
|
|
elif self.file_type == filetypes.pipfile_lock:
|
|
updater_class = updater.PipfileLockUpdater
|
|
elif self.file_type == filetypes.setup_cfg:
|
|
updater_class = updater.SetupCFGUpdater
|
|
else:
|
|
raise NotImplementedError
|
|
|
|
dep = Dependency(
|
|
name=self.name,
|
|
specs=self.specs,
|
|
line=self.line,
|
|
line_numbers=[
|
|
self.lineno,
|
|
]
|
|
if self.lineno != 0
|
|
else None,
|
|
dependency_type=self.file_type,
|
|
hashes=self.hashes,
|
|
extras=self.extras,
|
|
)
|
|
hashes = []
|
|
if self.hashes and update_hashes:
|
|
hashes = self.get_hashes(version)
|
|
|
|
return updater_class.update(
|
|
content=content, dependency=dep, version=version, hashes=hashes, spec="=="
|
|
)
|
|
|
|
@classmethod
|
|
def parse(
|
|
cls, s: str, lineno: int, file_type: str = filetypes.requirements_txt
|
|
) -> "Requirement":
|
|
"""
|
|
Parses a requirement from a line of text.
|
|
|
|
Args:
|
|
s (str): The line of text.
|
|
lineno (int): The line number.
|
|
file_type (str): The type of the file containing the requirement.
|
|
|
|
Returns:
|
|
Requirement: The parsed requirement.
|
|
"""
|
|
# setuptools requires a space before the comment. If this isn't the case, add it.
|
|
if "\t#" in s:
|
|
(parsed,) = parse_requirements(s.replace("\t#", "\t #"))
|
|
else:
|
|
(parsed,) = parse_requirements(s)
|
|
|
|
return cls(
|
|
name=parsed.name,
|
|
specs=parsed.specifier,
|
|
line=s,
|
|
lineno=lineno,
|
|
extras=list(parsed.extras),
|
|
file_type=file_type,
|
|
)
|