Chart each site's risk over time, one line per telemetry source

A site is often watched by more than one scanner: the built-in dynamic scan
and, where it is set up, Rapid7 InsightAppSec, with older runs recorded by
hand.  Those readings measure different things and should not be drawn as one
line, so the site page now shows a separate risk trend for each source, built
only once there is a finished scan to plot.

Each point is one scan, oldest on the left, joined into a line.  Its height is
a weighted risk worked out from the scan's own finding counts, with the same
per severity weights the grade uses, so every kind of scan lands on one scale
even when only counts came back.  The dot takes the color of the scan's worst
finding, and a small bubble above each point carries the scan id so a reading
is easy to pick out; the whole point links to that scan, and hovering it names
the date, the risk and the counts behind it.

The geometry is worked out in the shared chart helper and drawn as inline SVG,
so it needs no page script and reads the same in either theme.

Author: hackrange

app/core/charts.py

@@ -171,6 +171,93 @@ def y_at(value):
     }
 
 
+def risk_line(points: list, width: int = 720, height: int = 240,
+              pad_left: int = 40, pad_right: int = 16,
+              pad_top: int = 34, pad_bottom: int = 38) -> dict:
+    """-> a single risk line over a run of scans, already positioned.
+
+    One point per scan, left to right oldest to newest, joined by a line.  Each
+    point carries the scan's id so the page can print it as a small bubble and
+    link it, and the point's worst severity so the dot takes that color.  The
+    bubbles are staggered above and below the line in turn, so a run of close
+    scans does not stack their numbers on top of each other.
+
+    A single scan is drawn as one dot in the middle, not an invisible point,
+    because one reading is a reasonable thing to be looking at.
+
+    Each point needs: value (the weighted risk, higher is worse), id, label (a
+    short x-axis date), date (a full hover date), worst (a severity name or
+    "none"), counts (a severity->n dict), grade (a letter or None).
+    """
+    if not points:
+        return {"empty": True, "width": width, "height": height}
+
+    plot_w = width - pad_left - pad_right
+    plot_h = height - pad_top - pad_bottom
+
+    values = [max(0, int(p.get("value") or 0)) for p in points]
+    top = _nice_ceiling(max(values) if values else 0)
+    count = len(points)
+
+    def x_at(index):
+        if count == 1:
+            return pad_left + plot_w / 2.0
+        return pad_left + (plot_w * index / (count - 1.0))
+
+    def y_at(value):
+        return pad_top + plot_h - (plot_h * min(value, top) / float(top))
+
+    grid = []
+    for step in range(GRID_STEPS + 1):
+        value = top * step // GRID_STEPS
+        grid.append({"value": value, "y": round(y_at(value), 1)})
+
+    line = " ".join("%.1f,%.1f" % (x_at(i), y_at(values[i])) for i in range(count))
+
+    marks = []
+    for i, point in enumerate(points):
+        x, y = x_at(i), y_at(values[i])
+        label = str(point.get("id") or "")
+        bubble_w = round(len(label) * 6.6 + 12, 1)
+        # Centre the bubble on the point, then keep it inside the frame so the
+        # first and last ones are not clipped at the edges.
+        bx = x - bubble_w / 2.0
+        bx = max(2.0, min(bx, width - bubble_w - 2.0))
+        above = (i % 2 == 0)             # alternate sides to cut overlap
+        by = (y - 26) if above else (y + 8)
+        by = max(2.0, min(by, height - 18.0))
+        marks.append({
+            "x": round(x, 1), "y": round(y, 1),
+            "id": point.get("id"), "value": values[i],
+            "worst": point.get("worst") or "none",
+            "date": point.get("date", ""), "grade": point.get("grade"),
+            "counts": point.get("counts", {}), "label": label,
+            "bubble_x": round(bx, 1), "bubble_y": round(by, 1),
+            "bubble_w": bubble_w, "text_x": round(bx + bubble_w / 2.0, 1),
+            "text_y": round(by + 11.5, 1),
+        })
+
+    every = max(1, count // 8)
+    wanted = [i for i in range(count) if i % every == 0]
+    if count > 1:
+        if wanted and (count - 1 - wanted[-1]) <= max(1, every // 2):
+            wanted.pop()
+        wanted.append(count - 1)
+    labels = []
+    for i in sorted(set(wanted)):
+        anchor = "start" if i == 0 else "end" if i == count - 1 else "middle"
+        labels.append({"x": round(x_at(i), 1), "anchor": anchor,
+                       "text": points[i].get("label", "")})
+
+    return {
+        "empty": False, "width": width, "height": height,
+        "pad_left": pad_left, "pad_top": pad_top,
+        "plot_w": plot_w, "plot_h": plot_h, "top": top,
+        "line": line, "grid": grid, "marks": marks, "labels": labels,
+        "first": values[0], "last": values[-1], "change": values[-1] - values[0],
+    }
+
+
 def heat_levels(rows: list, key: str = "total") -> list:
     """Add a 0 to 4 shade to each row, by where it sits against the busiest.
 

app/core/external.py

@@ -1709,6 +1709,83 @@ def findings_of(scan: ExternalDastScan) -> list:
         return []
 
 
+# The telemetry sources a site's scans can come from, grouped for the risk
+# trend.  The key decides which scans share a chart; the label titles it.
+_SOURCE_ORDER = {"builtin": 0, "insightappsec": 1, "remote": 2, "recorded": 3}
+
+
+def _scan_source(scan: ExternalDastScan) -> tuple:
+    """(group key, shown label) for one scan's telemetry source.  Rapid7 scans
+    come in as source "remote" with the InsightAppSec provider, so they are
+    split off from any other remote server by the provider."""
+    if (scan.provider or "") == "insightappsec":
+        return "insightappsec", "Rapid7 InsightAppSec"
+    if scan.source == "builtin":
+        return "builtin", "Built-in DAST"
+    if scan.source == "remote":
+        return "remote", (scan.server_name or "DAST server")
+    if scan.source == "recorded":
+        return "recorded", "Recorded"
+    return (scan.source or "other"), SCAN_SOURCES.get(
+        scan.source, (scan.source or "Other").title())
+
+
+def scan_risk_value(scan: ExternalDastScan) -> int:
+    """One weighted risk number from a scan's severity counts, so every scan -
+    a full built-in report, a counts-only InsightAppSec import, a recorded one
+    - lands on the same scale.  The weights are the report's own per-finding
+    demerits, so a point here means what a point in the grade means: higher is
+    worse."""
+    from app.core.dast_report import DEMERIT
+
+    return sum(DEMERIT.get(s, 0) * (getattr(scan, s) or 0) for s in SEVERITIES)
+
+
+def _scan_when_ts(scan: ExternalDastScan) -> float:
+    """A sortable timestamp for a scan, by its own date then when it started,
+    tolerant of a naive datetime coming back from SQLite."""
+    when = scan.scanned_at or scan.started_at
+    if when is None:
+        return 0.0
+    if when.tzinfo is None:
+        when = when.replace(tzinfo=timezone.utc)
+    return when.timestamp()
+
+
+def risk_trends(db, asset_id: int) -> list:
+    """One risk-over-time line per telemetry source for a site.  A site scanned
+    by both the built-in DAST and Rapid7 gets two charts, one each: the two
+    measure different things and must not be drawn as a single line.  Only
+    finished scans carry final counts, so only those become points."""
+    from app.core import charts
+
+    scans = (db.query(ExternalDastScan)
+             .filter(ExternalDastScan.asset_id == asset_id,
+                     ExternalDastScan.status == "done")
+             .all())
+    groups: dict = {}
+    for s in scans:
+        key, label = _scan_source(s)
+        groups.setdefault(key, {"label": label, "scans": []})["scans"].append(s)
+    out = []
+    for key in sorted(groups, key=lambda k: (_SOURCE_ORDER.get(k, 9), k)):
+        g = groups[key]
+        ordered = sorted(g["scans"], key=lambda s: (_scan_when_ts(s), s.id))
+        points = []
+        for s in ordered:
+            when = s.scanned_at or s.started_at
+            points.append({
+                "value": scan_risk_value(s), "id": s.id,
+                "label": when.strftime("%d %b") if when else "",
+                "date": when.strftime("%Y-%m-%d %H:%M") if when else "",
+                "worst": s.severity or "none",
+                "counts": {sev: getattr(s, sev) or 0 for sev in SEVERITIES},
+            })
+        out.append({"key": key, "label": g["label"],
+                    "count": len(points), "chart": charts.risk_line(points)})
+    return out
+
+
 def _iso(value):
     if value is None:
         return None

app/web/routes_external.py

@@ -729,6 +729,7 @@ def detail_page(asset_id: int, request: Request, scans: int = 1,
             picks_json=json.dumps(external.picks_of(db, asset)),
             environments=external.environments(db), statuses=external.STATUSES,
             scan_views=[external.scan_dict(s) for s in page_scans],
+            risk_trends=external.risk_trends(db, asset.id),
             servers=db.query(DastServer).order_by(DastServer.name).all(),
             ias=_ias_choices(db, asset),
             profiles=dast_engines.PROFILES, active=dast_engines.ACTIVE_PROFILES,

app/web/static/app.css

@@ -753,6 +753,28 @@ table.audit-detail td { padding: .2rem 0; font-size: .9rem; }
 .chart .boundary { stroke: var(--text-faint); stroke-width: 1;
   stroke-dasharray: 3 3; }
 
+/* The risk trend: one line per telemetry source, a dot per scan colored by its
+   worst finding, and a small bubble carrying the scan id above each point. */
+.risk-trend { margin-bottom: .6rem; }
+.risk-trend + .risk-trend { border-top: 1px solid var(--border);
+  padding-top: .5rem; }
+.risk-trend-title { font-weight: 600; font-size: .9rem; margin: .1rem 0 0 .15rem; }
+.risk-trend-note { margin: .15rem .25rem 0; }
+.chart .riskline { fill: none; stroke: var(--info); stroke-width: 2;
+  stroke-linejoin: round; stroke-linecap: round; }
+.chart .dot { stroke: var(--bg); stroke-width: 1.5; }
+.chart .dot.band-critical { fill: var(--bad); opacity: 1; }
+.chart .dot.band-high     { fill: var(--bad); opacity: .72; }
+.chart .dot.band-medium   { fill: var(--warn); opacity: 1; }
+.chart .dot.band-low      { fill: var(--info); opacity: 1; }
+.chart .dot.band-info     { fill: var(--text-faint); opacity: 1; }
+.chart .dot.band-clean    { fill: var(--ok); opacity: 1; }
+.chart a:hover .dot { stroke: var(--text); }
+.chart .bubble { fill: var(--bg); stroke: var(--border-strong); stroke-width: 1; }
+.chart a:hover .bubble { stroke: var(--info); }
+.chart .bubble-text { fill: var(--text-soft); font-size: 10px; font-weight: 600; }
+.chart a:hover .bubble-text { fill: var(--info); }
+
 .legend { display: flex; flex-wrap: wrap; gap: .35rem 1rem;
   padding: .6rem 1rem 0; font-size: .76rem; color: var(--text-soft); }
 .legend .key { display: inline-flex; align-items: center; gap: .35rem; }

app/web/templates/external_site.html

@@ -230,6 +230,47 @@
       </div>
     </div>
 
+    <div class="card" data-live="site-risk-trend">
+      <div class="card-head">Risk trend
+        <span class="faint small" style="font-weight:400">one line per telemetry source, higher is worse</span></div>
+      {% if risk_trends %}
+      <div class="card-body">
+        {% for t in risk_trends %}
+        {% set c = t.chart %}
+        <div class="risk-trend">
+          <div class="risk-trend-title">{{ t.label }}
+            <span class="faint small" style="font-weight:400">{{ c.last }} now{% if c.marks|length > 1 %}, {{ c.change if c.change >= 0 else -c.change }} {{ 'higher' if c.change > 0 else ('lower' if c.change < 0 else 'unchanged') }} than the first of {{ t.count }} scans{% else %}, {{ t.count }} scan{% endif %}</span></div>
+          <div class="chart-wrap">
+            <svg class="chart risk-chart" viewBox="0 0 {{ c.width }} {{ c.height }}"
+                 preserveAspectRatio="xMidYMid meet" role="img"
+                 aria-label="Risk trend for {{ t.label }}">
+              {% for line in c.grid %}
+              <line class="grid" x1="{{ c.pad_left }}" y1="{{ line.y }}" x2="{{ c.width - 12 }}" y2="{{ line.y }}"/>
+              <text class="axis" x="{{ c.pad_left - 6 }}" y="{{ line.y + 4 }}" text-anchor="end">{{ line.value }}</text>
+              {% endfor %}
+              {% if c.marks|length > 1 %}
+              <polyline class="riskline" points="{{ c.line }}"/>
+              {% endif %}
+              {% for m in c.marks %}
+              <a href="{{ base }}/external/{{ asset.id }}/scans/{{ m.id }}">
+                <title>Scan {{ m.id }}{% if m.date %} &middot; {{ m.date }}{% endif %} &middot; risk {{ m.value }}{% for sev in ['critical','high','medium','low','info'] %}{% if m.counts[sev] %} &middot; {{ m.counts[sev] }} {{ sev }}{% endif %}{% endfor %}</title>
+                <circle class="dot band-{{ m.worst if m.worst != 'none' else 'clean' }}" cx="{{ m.x }}" cy="{{ m.y }}" r="4"/>
+                <rect class="bubble" x="{{ m.bubble_x }}" y="{{ m.bubble_y }}" width="{{ m.bubble_w }}" height="16" rx="8"/>
+                <text class="bubble-text" x="{{ m.text_x }}" y="{{ m.text_y }}" text-anchor="middle">{{ m.label }}</text>
+              </a>
+              {% endfor %}
+              {% for label in c.labels %}
+              <text class="axis" x="{{ label.x }}" y="{{ c.height - 8 }}" text-anchor="{{ label.anchor }}">{{ label.text }}</text>
+              {% endfor %}
+            </svg>
+          </div>
+        </div>
+        {% endfor %}
+        <div class="faint small risk-trend-note">Each point is one finished scan, oldest on the left.  Risk weighs findings by severity: critical 25, high 12, medium 5, low 2.  The number in each bubble is the scan id; click a point to open that scan.  A dot takes the color of the scan's worst finding.</div>
+      </div>
+      {% else %}<div class="empty">No finished scan yet, so there is nothing to trend.</div>{% endif %}
+    </div>
+
     <div class="card" data-live="site-scans">
       <div class="card-head">Dynamic scans
         {% if scan_total %}<span class="faint small" style="font-weight:400">{{ scan_total }} in all, newest first</span>{% endif %}</div>