- add download and mirror cache mechanism

- add `--download-only` option to `aura scan`
- various bugfixes

Author: RootLUG

aura/analyzers/python/nodes.py

@@ -6,6 +6,7 @@
 import os
 import typing
 import inspect
+import logging
 from abc import ABCMeta, abstractmethod
 from enum import Enum
 from pathlib import Path
@@ -26,6 +27,8 @@
     str,
     int,
 )
+logger = logging.getLogger(__name__)
+
 
 @total_ordering
 class Taints(Enum):
@@ -231,12 +234,15 @@ def pprint(self):
         from prettyprinter import pprint as pp
         pp(self)
 
-    def add_taint(self, taint: Taints, context: Context, lock=False, taint_log=None) -> bool:
+    def add_taint(self, taint: Taints, context: Context, lock=False, taint_log: typing.Optional[TaintLog]=None) -> bool:
         """
         Assign a taint to the node
         Operation is ignored if the current taint is already higher or equal
         return True if the taint was modified (increased)
         """
+        if taint_log:
+            logger.debug(f"{taint_log.message} at {taint_log.path}:{taint_log.line_no}")
+
         if lock:
             self._taint_locked = True
             if taint != self._taint_class:

aura/analyzers/python/readonly.py

@@ -17,7 +17,7 @@ def __init__(self, *, location: ScanLocation):
     def __call__(self) -> typing.Generator[Detection, None, None]:
         if not self.hooks:
             return
-        elif self.location.is_python_source_code:
+        elif not self.location.is_python_source_code:
             return
         try:
             for x in self.hooks:

aura/cache.py

@@ -0,0 +1,90 @@
+import os
+import shutil
+import hashlib
+import logging
+from pathlib import Path
+from typing import Optional
+
+from . import utils
+from . import config
+
+logger = logging.getLogger(__name__)
+
+
+class Cache:
+    DISABLE_CACHE = bool(os.environ.get("AURA_NO_CACHE"))
+    __location: Optional[Path] = None
+
+    @classmethod
+    def get_location(cls) -> Optional[Path]:
+        if cls.DISABLE_CACHE:
+            return None
+
+        if cls.__location is None:
+            c = os.environ.get("AURA_CACHE_LOCATION") or config.CFG["aura"].get("cache_location")
+            if c:
+                c = Path(c).expanduser().resolve()
+                logger.debug(f"Cache location set to {c}")
+
+                if not c.exists():
+                    c.mkdir(parents=True)
+                cls.__location = c
+
+        return cls.__location
+
+    @classmethod
+    def purge_cache(cls):  # TODO
+        total, used, free = shutil.disk_usage(cls.get_location())
+        cache_items = [x for x in cls.get_location().iterdir()]
+        cache_items.sort(key=lambda x: x.stat().st_mtime)
+
+    @classmethod
+    def proxy_url(cls, *, url, fd, cache_id=None):
+        if cls.get_location() is None:
+            return utils.download_file(url, fd=fd)
+
+        if cache_id is None:
+            cache_id = hashlib.md5(url).hexdigest()
+
+        cache_id = f"url_{cache_id}"
+        cache_pth: Path = cls.get_location()/cache_id
+
+        if cache_pth.is_file():
+            logger.info(f"Loading {cache_id} from cache")
+            with cache_pth.open("rb") as cfd:
+                shutil.copyfileobj(cfd, fd)
+                return
+
+        try:
+            with cache_pth.open("wb") as cfd:
+                utils.download_file(url, cfd)
+                cfd.flush()
+            with cache_pth.open("rb") as cfd:
+                shutil.copyfileobj(cfd, fd)
+        except Exception as exc:
+            cache_pth.unlink(missing_ok=True)
+            raise exc
+
+    @classmethod
+    def proxy_mirror(cls, *, src: Path, cache_id=None):
+        if not src.exists():
+            return None
+        elif cls.get_location() is None:
+            return src
+
+        if cache_id is None:
+            cache_id = src.name
+
+        cache_id = f"mirror_{cache_id}"
+        cache_pth: Path = cls.get_location() / cache_id
+
+        try:
+            if not cache_pth.exists():
+                with cache_pth.open("wb") as cfd:
+                    with src.open("rb") as fd:
+                        shutil.copyfileobj(fd, cfd)
+                        cfd.flush()
+            return cache_pth
+        except Exception as exc:
+            cache_pth.unlink(missing_ok=True)
+            raise exc

aura/cli.py

@@ -69,6 +69,7 @@ def cli(ctx, **kwargs):
 )
 @click.option("--async", "fork_mode", flag_value=True)
 @click.option("--no-async", "fork_mode", flag_value=False)
+@click.option("--download-only", "download_only", flag_value=True)
 def scan(
     uri,
     verbose=None,
@@ -78,7 +79,8 @@ def scan(
     benchmark=False,
     benchmark_sort="cumtime",
     filter_tags=None,
-    fork_mode=False
+    fork_mode=False,
+    download_only=False,
 ):
     output_opts = {
         "tags": filter_tags
@@ -109,7 +111,7 @@ def scan(
         cProfile, pstats, pr, io = None, None, None, None
 
     try:
-        commands.scan_uri(uri, metadata=meta)
+        commands.scan_uri(uri, metadata=meta, download_only=download_only)
     except exceptions.FeatureDisabled as e:
         LOGGER.exception(e.args[0], exc_info=e)
         click.secho(e.args[0], err=True, fg="red")

aura/commands.py

@@ -7,7 +7,7 @@
 import traceback
 from pathlib import Path
 from functools import partial
-from typing import Union, Optional, Tuple, Generator
+from typing import Union, Optional, Tuple, Generator, List
 
 import click
 from prettyprinter import pprint
@@ -42,8 +42,6 @@ def check_requirement(pkg):
             "format": "plain",
             "min_score": 0,
         }
-
-
         hits = []
 
         for location in handler.get_paths(metadata=metadata):
@@ -75,7 +73,7 @@ def scan_worker(item: ScanLocation) -> Generator[Detection, None, None]:
         yield from sandbox.run()
 
 
-def scan_uri(uri, metadata: Union[list, dict]=None) -> list:
+def scan_uri(uri, metadata: Union[list, dict]=None, download_only: bool=False) -> List[Detection]:
     with utils.enrich_exception(uri, metadata):
         start = time.time()
         handler = None
@@ -105,7 +103,10 @@ def scan_uri(uri, metadata: Union[list, dict]=None) -> list:
 
             # FIXME: metadata=metadata
             for x in handler.get_paths(metadata={"analyzers": metadata["analyzers"]}):  # type: ScanLocation
-                all_hits.extend(scan_worker(x))
+                if download_only:
+                    continue
+                else:
+                    all_hits.extend(scan_worker(x))
 
             for formatter in formatters:
                 try:

aura/data/aura_config.yaml

@@ -6,6 +6,9 @@ aura: &aura_config
   # https://docs.python.org/3.7/library/warnings.html#overriding-the-default-filter
   warnings: default
 
+  # Directory location for caching, unset to disable caching
+  cache_location: &cache_location ~/.aura_cache
+
   # Path to the location of the offline PyPI mirror
   # This option can be safely disabled if you don't have a local mirror, some advanced features require this
   mirror: /var/pypi_mirror/pypi/web/

aura/diff.py

@@ -57,6 +57,13 @@ class Diff():
 
     def __post_init__(self):
         assert self.operation in ("A", "D", "M", "R")
+
+        if self.a_mime == "text/x-script.python":
+            self.a_mime = "text/x-python"
+
+        if self.b_mime == "text/x-script.python":
+            self.b_mime = "text/x-python"
+
         # a_path, b_path = self.a_path, self.b_path
         #
         # if a_path and b_path:

aura/output/text.py

@@ -1,4 +1,5 @@
 import re
+import os
 import sys
 from shutil import get_terminal_size
 from dataclasses import dataclass
@@ -55,6 +56,8 @@
 OK = '\u2713'
 NOK = '\u2717'
 
+TTY_COLORS = bool(os.environ.get("AURA_FORCE_COLORS", False)) or None
+
 
 
 class PrettyReport:
@@ -81,7 +84,7 @@ def ansi_length(cls, line:str):
         return len(cls.ANSI_RE.sub("", line))
 
     def print_separator(self, sep="\u2504", left="\u251C", right="\u2524"):
-        secho(f"{left}{sep*(self.width-2)}{right}", file=self.fd)
+        secho(f"{left}{sep*(self.width-2)}{right}", file=self.fd, color=TTY_COLORS)
 
     def print_thick_separator(self):
         self.print_separator(left="\u255E", sep="\u2550", right="\u2561")
@@ -96,11 +99,11 @@ def print_heading(self, text, left="\u251C", right="\u2524", infill="\u2591"):
         text_len = self.ansi_length(text)
         ljust = (self.width-4-text_len)//2
         rjust = self.width-4-text_len-ljust
-        secho(f"{left}{infill*ljust} {text} {infill*rjust}{right}", file=self.fd)
+        secho(f"{left}{infill*ljust} {text} {infill*rjust}{right}", file=self.fd, color=TTY_COLORS)
 
     def align(self, line, pos=-1, left="\u2502 ", right=" \u2502"):
         line = self._align_text(line, self.width - len(left) - len(right), pos=pos)
-        secho(f"{left}{line}{right}", file=self.fd)
+        secho(f"{left}{line}{right}", file=self.fd, color=TTY_COLORS)
 
     def wrap(self, text, left="\u2502 ", right=" \u2502"):
         remaining_len=self.width - len(left) - len(right)
@@ -271,7 +274,7 @@ def output(self, hits, scan_metadata: dict):
         if score < self.min_score:
             return
 
-        secho("\n", file=self._fd)  # Empty line for readability
+        secho("\n", file=self._fd, color=TTY_COLORS)  # Empty line for readability
         self._formatter.print_top_separator()
         self._formatter.print_heading(style(f"Scan results for {scan_metadata.get('name', 'N/A')}", fg="bright_green"))
         score_color = "bright_green" if score == 0 else "bright_red"

aura/package.py

@@ -21,6 +21,7 @@
 
 from . import config
 from . import github
+from . import cache
 from .json_proxy import loads
 from . import utils
 from . import exceptions
@@ -109,7 +110,7 @@ def download_release(
 
         for url in filtered:
             with open(dest / url["filename"], "wb") as fd:
-                utils.download_file(url["url"], fd)
+                cache.Cache.proxy_url(url=url["url"], fd=fd, cache_id=url["filename"])
             files.append(url)
 
         return files

aura/pattern_matching.py

@@ -4,6 +4,7 @@
 import inspect
 import fnmatch
 import string
+import logging
 from abc import ABCMeta, abstractmethod
 from typing import List, Mapping
 from functools import lru_cache
@@ -16,6 +17,7 @@
 
 
 PATTERN_CACHE = None
+logger = logging.getLogger(__name__)
 
 
 class PatternMatcher(metaclass=ABCMeta):
@@ -272,6 +274,7 @@ class AnyOf(list):
 
 
     def __init__(self, signature: dict):
+        self._id = None
         self._signature = signature
 
         if type(self._signature["pattern"]) == str:
@@ -290,16 +293,21 @@ def _compile_src(cls, src: str) -> nodes.NodeType:
 
     @property
     def id(self) -> str:
-        if "id" in self._signature:
-            return self._signature["id"]
+        if self._id is None:
+            if "id" in self._signature:
+                return self._signature["id"]
 
-        chars = string.ascii_letters + string.digits + "._-"
-        return "".join(x for x in self._signature["pattern"] if x in chars)
+            chars = string.ascii_letters + string.digits + "._-"
+            self._id = "".join(x for x in self._signature["pattern"] if x in chars)
+
+        return self._id
 
     def match(self, node: nodes.ASTNode) -> bool:
         return self.ctx.match(node, self._compiled)
 
     def apply(self, context: nodes.Context):
+        logger.debug(f"Applying AST pattern {self.id} at {context.node.line_no}")
+
         if "tags" in self._signature:
             context.node.tags |= set(self._signature["tags"])