-refactor typosquatting for better text output

- various bugfixes and code enhancements

Author: RootLUG

aura/analyzers/data_finder.py

@@ -32,11 +32,10 @@ def __init__(self, *args, **kwargs):
 
         self._no_blobs = False
 
-        if "AURA_NO_BLOB" in os.environ:
+        if "AURA_NO_BLOBS" in os.environ:
             self._no_blobs = True
 
         self.__min_blob_size = self.get_min_size()
-        pass
 
     @classmethod
     def get_min_size(cls) -> int:

aura/cache.py

@@ -43,7 +43,12 @@ def proxy_url(cls, *, url, fd, cache_id=None):
             return utils.download_file(url, fd=fd)
 
         if cache_id is None:
-            cache_id = hashlib.md5(url).hexdigest()
+            if isinstance(url, str):
+                burl = url.encode()
+            else:
+                burl = url
+
+            cache_id = hashlib.md5(burl).hexdigest()
 
         cache_id = f"url_{cache_id}"
         cache_pth: Path = cls.get_location()/cache_id

aura/cli.py

@@ -223,12 +223,13 @@ def update_aura():
 @click.option("-o", "--out", default="-", type=click.File("w"))
 @click.option("-m", "--max-distance", default=2, type=click.IntRange(min=0, max=10))
 @click.option("-l", "--limit", default=100, type=click.INT)
-def find_typosquatting(out, max_distance, limit=100):
+@click.argument("pkg", nargs=-1)
+def find_typosquatting(out, max_distance, limit=100, pkg=None):
     if limit <= 0:
         click.secho("Invalid value for limit", file=sys.stderr)
         sys.exit(1)
 
-    commands.generate_typosquatting(out=out, distance=max_distance, limit=limit)
+    commands.generate_typosquatting(out=out, distance=max_distance, limit=limit, pkgs=pkg)
 
 
 @cli.command()

aura/commands.py

@@ -214,13 +214,28 @@ def show_info():
     formatter.output_info_data(info_data)
 
 
-def generate_typosquatting(out, distance=2, limit=None):
+def generate_typosquatting(out, distance=2, limit=None, pkgs=None):
+    from .output.text import PrettyReport
+
+    p = PrettyReport()  # TODO: convert into plugin system
+
+    if not pkgs:
+        pkgs = typos.get_popular_packages()
+
     f = partial(typos.damerau_levenshtein, max_distance=distance)
-    for num, (x, y) in enumerate(typos.enumerator(typos.generate_popular(), f)):
-        if limit and num >= limit:
+    combinations = typos.generate_combinations(left=pkgs)
+
+    for idx, data in enumerate(typos.enumerator(combinations, f)):
+        if limit and idx >= limit:
             break
 
-        out.write(json.dumps({"original": x, "typosquatting": y}) + "\n")
+        try:
+            diff_table = data["orig_pkg"]._cmp_info(data["typo_pkg"])
+            t1 = data["orig_score"].get_score_table()
+            t2 = data["typo_score"].get_score_table()
+            p.print_tables(t1, t2, diff_table)
+        except exceptions.NoSuchPackage:
+            continue
 
 
 def prefetch(*uris):

aura/diff.py

@@ -8,7 +8,7 @@
 import tempfile
 import shutil
 import pprint
-from typing import Union, Optional, List
+from typing import Union, Optional, List, Tuple
 from pathlib import Path
 from collections import defaultdict
 from dataclasses import dataclass
@@ -358,6 +358,8 @@ def pprint(self):
 
 
 class DiffDetections:
+    ANALYSIS_CACHE = {}
+
     def __init__(self, file_diffs: List[Diff], a_location: ScanLocation, b_location: ScanLocation):
         self.file_diffs: List[Diff] = file_diffs
         self.a_loc: ScanLocation = a_location
@@ -384,9 +386,12 @@ def __init__(self, file_diffs: List[Diff], a_location: ScanLocation, b_location:
             b_detections = b_pairs.get(d.b_ref, [])
             d.add_detections(a_detections, b_detections)
 
-    def scan_location(self, location):
-        sandbox = Analyzer(location=location)
-        return tuple(sandbox.run())
+    def scan_location(self, location) -> Tuple[Detection]:
+        if location.location not in self.ANALYSIS_CACHE:
+            sandbox = Analyzer(location=location)
+            detections = tuple(sandbox.run())
+            self.ANALYSIS_CACHE[location.location] = detections
+        return self.ANALYSIS_CACHE[location.location]
 
     def pair_hits(self, diff_refs, hits):
         orphans = []

aura/output/table.py

@@ -1,3 +1,5 @@
+from click import style
+
 from dataclasses import dataclass, field
 from typing import Any, List, Iterable, Optional
 
@@ -33,6 +35,10 @@ def style(self) -> dict:
 
         return s
 
+    @property
+    def pretty(self) -> str:
+        return style(str(self), **self.style)
+
     def asdict(self) -> dict:
         data = {
             "value": self.value
@@ -81,6 +87,11 @@ def __iadd__(self, other):
     def __iter__(self):
         yield from self.rows
 
+    @property
+    def width(self) -> int:
+        title_width = len(self.metadata.get("title", ""))
+        return max(sum(self.col_len)+3, title_width)
+
     def asdict(self) -> dict:
         d = {"rows": [
             [c.asdict() for c in row] for row in self.rows
@@ -89,3 +100,27 @@ def asdict(self) -> dict:
             d["metadata"] = self.metadata
 
         return d
+
+    def pprint(self, preport=None):
+        from .text import PrettyReport
+
+        if preport is None:
+            preport = PrettyReport()
+
+        preport.print_tables(self, self)
+
+        preport.print_top_separator()
+
+        if self.metadata.get("title"):
+            preport.print_heading(self.metadata["title"])
+
+        for row in self:
+            cols = []
+            for idx, col in enumerate(row):
+                text = preport._align_text(style(str(col), **col.style), width=self.col_len[idx])
+                cols.append(text)
+
+            preport.align(" \u2502 ".join(cols))
+
+        preport.print_bottom_separator()
+

aura/output/text.py

@@ -1,6 +1,7 @@
 import re
 import os
 import sys
+import itertools
 from shutil import get_terminal_size
 from dataclasses import dataclass
 from textwrap import wrap
@@ -69,40 +70,54 @@ class PrettyReport:
     """, re.VERBOSE)
 
     def __init__(self, fd=None):
-        width = config.get_settings("aura.text-output-width", "auto")
         self.term_width = get_terminal_size(fallback=(120, 24))[0]
 
-        if width == "auto":
-            self.width = self.term_width
+        if "AURA_TERM_WIDTH" in os.environ:
+            self.width = int(os.environ["AURA_TERM_WIDTH"])
         else:
-            self.width = int(width or 120)
+            width = config.get_settings("aura.text-output-width", "auto")
+            if width == "auto":
+                self.width = self.term_width
+            else:
+                self.width = int(width or 120)
 
         self.fd = fd
 
     @classmethod
     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, color=TTY_COLORS)
+    def print_separator(self, sep="\u2504", left="\u251C", right="\u2524", width=None):
+        if width is None:
+            width = self.width
+
+        secho(f"{left}{sep*(width-2)}{right}", file=self.fd, color=TTY_COLORS)
 
     def print_thick_separator(self):
         self.print_separator(left="\u255E", sep="\u2550", right="\u2561")
 
-    def print_top_separator(self):
-        self.print_separator(left="\u2552", sep="\u2550", right="\u2555")
+    def print_top_separator(self, **kwargs):
+        self.print_separator(left="\u2552", sep="\u2550", right="\u2555", **kwargs)
 
-    def print_bottom_separator(self):
-        self.print_separator(left="\u2558", sep="\u2550", right="\u255B")
+    def print_bottom_separator(self, **kwargs):
+        self.print_separator(left="\u2558", sep="\u2550", right="\u255B", **kwargs)
+
+    def generate_heading(self, text, left="\u251C", right="\u2524", infill="\u2591", width=None):
+        if width is None:
+            width = self.width - len(left) - len(right) - 2
 
-    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, color=TTY_COLORS)
+        ljust = (width - text_len) // 2
+        rjust = width - text_len - ljust
+        return f"{left}{infill*ljust} {text} {infill*rjust}{right}"
+
+    def print_heading(self, *args, **kwargs):
+        secho(self.generate_heading(*args, **kwargs), 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)
+    def align(self, line, pos=-1, left="\u2502 ", right=" \u2502", width=None):
+        if width is None:
+            width = self.width
+        line = self._align_text(line, width - len(left) - len(right), pos=pos)
         secho(f"{left}{line}{right}", file=self.fd, color=TTY_COLORS)
 
     def wrap(self, text, left="\u2502 ", right=" \u2502"):
@@ -136,6 +151,34 @@ def _align_text(self, text, width, pos=-1):
         else:
             return " " * (remaining_len - content_len) + text
 
+    def print_tables(self, *tables):
+        table_widths = [t.width+2 for t in tables]
+
+        self.print_top_separator()
+
+        titles = [t.metadata.get("title", "N/A") for t in tables]
+        tparts = [tuple(self.generate_heading(title, width=w, left="", right="") for w, title in zip(table_widths, titles))]
+
+        for idx, rows in enumerate(itertools.zip_longest(*tables, fillvalue="")):
+
+            full_row = []
+
+            for ridx, row in enumerate(rows):
+                text = " \u2506 ".join(self._align_text(c.pretty, width=tables[ridx].col_len[cidx]) for cidx, c in enumerate(row))
+                text = self._align_text(text, width=table_widths[ridx]+2)
+
+                full_row.append(text)
+
+            tparts.append(full_row)
+
+        for idx, tpart in enumerate(tparts):
+            self.align(" \u2551 ".join(tpart))
+            if idx == 0:
+                self.print_thick_separator()
+
+        self.print_bottom_separator()
+
+
 
 @dataclass()
 class TextBase:
@@ -224,7 +267,7 @@ def imports_to_tree(self, items: list) -> dict:
 
         return root
 
-    def output_table(self, table):
+    def output_table(self, table: Table):
         out = PrettyReport(fd=self._fd)
         out.print_top_separator()
 
@@ -405,8 +448,8 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     def output_diff(self, diff_analyzer):
         out = PrettyReport(fd=self._fd)
 
-        for table in diff_analyzer.tables:
-            self.output_table(table)
+        if diff_analyzer.tables:
+            out.print_tables(*diff_analyzer.tables)
 
         for diff in self.filtered(diff_analyzer.diffs):
             out.print_separator(left="\u2552", sep="\u2550", right="\u2555")