Automatische Quellcodepositionen mit Rego

Bei Snyk sind wir große Fans von Rego von Open Policy Agent. Snyk IaC basiert auf einer umfangreichen Sammlung von Regeln, die in Rego geschrieben sind. Kunden können außerdem eigene benutzerdefinierte Regeln hinzufügen.

Kürzlich haben wir eine Reihe von Verbesserungen an Snyk IaC veröffentlicht. In diesem Blogbeitrag werfen wir einen technischen Blick auf ein besonders interessantes Feature: automatische Quellcodepositionen für Regelverstöße.

Bei der Prüfung von IaC-Dateien auf bekannte Probleme zeigt der aktualisierte Befehl `snyk iac test` für jeden Regelverstoß präzise Angaben zu Datei, Zeile und Spalte an. Das funktioniert auch für benutzerdefinierte Regeln, ohne dass Sie etwas dafür tun müssen.

Bevor wir einen eigenständigen Proof of Concept für diese Technik bereitstellen, müssen wir jedoch einige Vereinfachungen vornehmen. Die vollständige Implementierung finden Sie in unserer vereinheitlichten Policy-Engine.

Sehen wir uns zunächst ein CloudFormation-Beispiel an. Unsere IaC-Engine unterstützt zwar viele Formate und legt einen starken Schwerpunkt auf Terraform, aber CloudFormation eignet sich gut als Beispiel, da wir es ohne allzu viele Abhängigkeiten parsen können (schließlich ist es einfach YAML).

1Resources:
2  Vpc:
3    Type: AWS::EC2::VPC
4    Properties:
5      CidrBlock: 10.0.0.0/16
6  PublicSubnet:
7    Type: AWS::EC2::Subnet
8    Properties:
9      VpcId:
10        Ref: Vpc
11      CidrBlock: 10.0.0.0/24
12  PrivateSubnet:
13    Type: AWS::EC2::Subnet
14    Properties:
15      VpcId:
16        Ref: Vpc
17      CidrBlock: 10.0.128.0/20

Wir möchten sicherstellen, dass kein Subnetz einen CIDR-Block verwendet, der größer als `/24` ist. Schreiben wir also eine Rego-Policy, die genau das sicherstellt:

1package policy
2
3deny[resourceId] {
4    resource := input.Resources[resourceId]
5    resource.Type == "AWS::EC2::Subnet"
6    [_, mask] = split(resource.Properties.CidrBlock, "/")
7    to_number(mask) < 24
8}

Auf diese Weise gibt `deny` eine Menge abgelehnter Ressourcen zurück. Wir gehen nicht näher darauf ein, wie Rego funktioniert. Wenn Sie mehr erfahren möchten, empfehlen wir den hervorragenden Kurs OPA by Example.

Wir können das Problem in zwei Teile aufteilen:

1. Wir müssen ableiten, dass unsere Policy das Attribut `CidrBlock` verwendet.

2. Anschließend rufen wir die Position im Quellcode ab.

Beginnen wir mit (2), denn so können wir uns gut mit dem Code vertraut machen.

Quellcodeposition abrufen

Eine Quellcodeposition sieht so aus:

1type Location struct {
2    File   string
3    Line   int
4    Column int
5}
6func (loc Location) String() string {
7    return fmt.Sprintf("%s:%d:%d", loc.File, loc.Line, loc.Column)
8}

Außerdem führen wir einen Hilfstyp ein, der Pfade in YAML darstellt. In YAML gibt es zwei Arten verschachtelter Dokumente: Arrays und Objekte.

1some_array:
2- hello
3- world
4some_object:
5  foo: bar

Um auf jedes beliebige Teildokument verweisen zu können, könnten wir etwas Ähnliches wie JSON-Pfade verwenden. Im obigen Beispiel würde [`"some_array", 1`] dann auf `"word"` verweisen. Da wir Arrays in unserem Proof of Concept jedoch nicht unterstützen, reicht ein Array aus Strings aus.

type Path []string

Ein Pfad könnte beispielsweise so aussehen:

Path{"Resources", "PrivateSubnet", "Properties", "CidrBlock"}

Jetzt können wir einen praktischen Typ bereitstellen, der YAML lädt und uns die `Location` bestimmter `Paths` mitteilt.

1type Source struct {
2    file string
3    root *yaml.Node
4}
5func NewSource(file string) (*Source, error) {
6    bytes, err := ioutil.ReadFile(file)
7    if err != nil {
8        return nil, err
9    }
10
11    var root yaml.Node
12    if err := yaml.Unmarshal(bytes, &root); err != nil {
13        return nil, err
14    }
15
16    return &Source{file: file, root: &root}, nil
17}

Die Quellcodeposition eines `Path` zu ermitteln, bedeutet, einen Baum aus YAML-Knoten zu durchlaufen:

1func (source *Source) Location(path Path) *Location {
2    cursor := source.root
3    for len(path) > 0 {
4        switch cursor.Kind {
5        // Ignore multiple docs in our PoC.
6        case yaml.DocumentNode:
7            cursor = cursor.Content[0]
8        case yaml.MappingNode:
9            // Objects are stored as an array.
10            // Content[2 * n] holds to the key and
11            // Content[2 * n + 1] to the value.
12            for i := 0; i < len(cursor.Content); i += 2 {
13                if cursor.Content[i].Value == path[0] {
14                    cursor = cursor.Content[i+1]
15                    path = path[1:]
16                }
17            }
18        }
19    }
20    return &Location{
21        File:   source.file,
22        Line:   cursor.Line,
23        Column: cursor.Column,
24    }
25}

Mengen und Bäume aus Pfaden

Damit haben wir das Problem vereinfacht: Statt automatisch die in einer Policy verwendeten Quellcodepositionen abzuleiten, müssen wir nun nur noch Attributpfade automatisch ableiten.

Das ist auch aus anderen Gründen wichtig: Snyk kann beispielsweise dieselben Policies sowohl auf IaC-Ressourcen als auch auf Ressourcen anwenden, die bei Cloud-Scans gefunden wurden. Letztere haben zwar keine wirklich aussagekräftigen Quellcodepositionen, aber durchaus aussagekräftige Attributpfade!

Wir möchten also Mengen von Attributpfaden definieren. Da Pfade auf Arrays basieren, können wir in Go leider nicht einfach `map[Path]struct{}` als Menge verwenden.

Stattdessen müssen wir sie in einem rekursiven Baum speichern.

type PathTree map[string]PathTree

Diese Darstellung hat weitere Vorteile. Im Allgemeinen interessieren uns nur die längsten Pfade, die eine Policy verwendet, da sie spezifischer sind. Unsere Beispiel-Policy verwendet sowohl `Path{"Resources", "PrivateSubnet", "Properties"}` als auch `Path{"Resources", "PrivateSubnet", "Properties", "CidrBlock"}` – uns interessiert nur der letztere.

Wir definieren eine rekursive Methode, die einen `Path` in unseren Baum einfügt:

1func (tree PathTree) Insert(path Path) {
2    if len(path) > 0 {
3        if _, ok := tree[path[0]]; !ok {
4            tree[path[0]] = map[string]PathTree{}
5        }
6        tree[path[0]].Insert(path[1:])
7    }
8}

…und außerdem eine Methode, mit der wir eine Liste von Paths abrufen können. Dabei wird etwas unnötig Speicher belegt, aber damit können wir leben.

1func (tree PathTree) List() []Path {
2    if len(tree) == 0 {
3        // Return the empty path.
4        return []Path{{}}
5    } else {
6        out := []Path{}
7        for k, child := range tree {
8            // Prepend `k` to every child path.
9            for _, childPath := range child.List() {
10                path := Path{k}
11                path = append(path, childPath...)
12                out = append(out, path)
13            }
14        }
15        return out
16    }
17}

Jetzt können wir die von einer Policy verwendeten `Paths` übersichtlich speichern und in Quellcodepositionen umwandeln.

Statische Analyse vs. Laufzeitanalyse

Als Nächstes müssen wir herausfinden, welche `Paths` in einer bestimmten Eingabe von einer Policy verwendet werden, und diese dann in den Baum `Insert`-en.

Das ist nicht einfach, da der Code die Eingabe auf verschiedene Weise verändern kann, bevor er die Pfade verwendet. Wir müssen uns sowohl benutzerdefinierte Funktionen ( `has_bad_subnet`) als auch integrierte Funktionen ( `object.get`) ansehen, um nur eines der möglichen Hindernisse zu veranschaulichen:

1has_bad_subnet(props) {
2    [_, mask] = split(props.CidrBlock, "/")
3    to_number(mask) < 24
4}
5
6deny[resourceId] {
7    resource := input.Resources[resourceId]
8    resource.Type == "AWS::EC2::Subnet"
9    has_bad_subnet(object.get(resource, "Properties", {}))
10}

Zum Glück sind wir damit nicht allein: Schon seit dem ersten Programm fragen sich Menschen, was Programme tun. Im Allgemeinen gibt es zwei Möglichkeiten, eine solche Frage zu einem Codeabschnitt zu beantworten:

• Statische Analyse: Wir versuchen, die Frage anhand des Syntaxbaums, der Typen und anderer statischer Informationen zu beantworten, die wir aus dem OPA-Interpreter abrufen (oder ihm hinzufügen) können. Der Vorteil: Wir müssen die Policy nicht ausführen. Das ist hilfreich, wenn wir den Autoren der Policy nicht vertrauen. Der Nachteil: Statische Analysemethoden führen in der Regel sowohl zu falsch negativen als auch zu falsch positiven Ergebnissen.

• Laufzeitanalyse: Wir zeichnen die Ausführung bestimmter Policies nach und leiten anhand der Laufzeitinformationen ab, welche `Paths` verwendet werden. Der Nachteil: Wir müssen die Policy tatsächlich ausführen, und diese Analyse kann die Auswertung der Policy verlangsamen.

Wir haben beide Ansätze ausprobiert, uns aber für den zweiten entschieden, da sich dieser deutlich zuverlässiger implementieren ließ und der Leistungsaufwand vernachlässigbar war. Außerdem ist es keine Entweder-oder-Entscheidung: Sie können auch einen hybriden Ansatz wählen und beide Methoden kombinieren.

OPA bietet eine Tracer-Schnittstelle, über die Ereignisse zu den Vorgängen im Interpreter empfangen werden können. Tracer werden häufig verwendet, um Metriken oder Debugging-Informationen an ein zentrales Protokoll zu senden. Heute setzen wir sie jedoch für einen anderen Zweck ein.

1type locationTracer struct {
2    tree PathTree
3}
4func newLocationTracer() *locationTracer {
5    return &locationTracer{tree: PathTree{}}
6}
7func (tracer *locationTracer) Enabled() bool {
8    return true
9}

Verwendung von Termen nachverfolgen

Rego ist eine ausdrucksstarke Sprache. Auch wenn durch Desugaring einiges in ein einfacheres Format für den Interpreter umgewandelt wird, gibt es immer noch eine ganze Reihe von Ereignissen.

Für uns sind nur zwei davon relevant. Ein Wert gilt als verwendet, wenn:

1. er mit einem anderen Ausdruck vereinheitlicht wird (Sie können sich das wie eine Zuweisung vorstellen; auf die Details gehen wir nicht ein), zum Beispiel:

x = input.Foo

Das gilt auch für `==` und `:=`. Da dieser Test fehlschlagen kann, können wir sagen, dass sowohl die linke als auch die rechte Seite verwendet wurden.

2. er als Argument an eine integrierte Funktion übergeben wird, zum Beispiel:

regex.match("/24$", input.cidr)

Rego übernimmt zwar einige Konzepte aus Lazy Languages, aber Argumente für integrierte Funktionen sind immer vollständig gebunden, bevor die Funktion aufgerufen wird. Daher können wir sagen, dass alle an die Funktion übergebenen Argumente verwendet wurden.

3. er als eigenständiger Ausdruck verwendet wird, zum Beispiel:

volume.encrypted

Das wird häufig verwendet, um Boolesche Werte auszuwerten und zu prüfen, ob Attribute vorhanden sind.

Jetzt ist es Zeit für die Implementierung. Wir gleichen zwei Ereignisse ab und übergeben sie jeweils an eine bestimmte Funktion, damit der Code besser lesbar ist:

1func (tracer *locationTracer) Trace(event *topdown.Event) {
2    switch event.Op {
3    case topdown.UnifyOp:
4        tracer.traceUnify(event)
5    case topdown.EvalOp:
6        tracer.traceEval(event)
7    }
8}

Das Einfügen in unseren `PathTree` behandeln wir später in einer Hilfsfunktion namens `used(*ast.Term)`. Zunächst markieren wir die linke und die rechte Seite der Vereinheitlichung als verwendet:

1func (tracer *locationTracer) traceUnify(event *topdown.Event) {
2    if expr, ok := event.Node.(*ast.Expr); ok {
3        operands := expr.Operands()
4        if len(operands) == 2 {
5            // Unification (1)
6            tracer.used(event.Plug(operands[0]))
7            tracer.used(event.Plug(operands[1]))
8        }
9    }
10}

`event.Plug` ist eine Hilfsfunktion, die Variablen durch ihre tatsächlichen Werte ersetzt.

Ein `EvalOp`-Ereignis deckt die oben genannten Fälle (2) und (3) ab. Bei einer integrierten Funktion erhalten wir ein Array von Termen. Das erste Element ist die Funktion, die übrigen Elemente sind die Argumente. Ob es sich um eine integrierte Funktion handelt, können wir anhand von `ast.BuiltinMap` prüfen.

Der Fall eines eigenständigen Ausdrucks ist einfach.

1func (tracer *locationTracer) traceEval(event *topdown.Event) {
2    if expr, ok := event.Node.(*ast.Expr); ok {
3        switch terms := expr.Terms.(type) {
4        case []*ast.Term:
5            if len(terms) < 1 {
6                // I'm not sure what this is, but it's definitely
7                // not a built-in function application.
8                break
9            }
10            operator := terms[0]
11            if _, ok := ast.BuiltinMap[operator.String()]; ok {
12                // Built-in function call (2)
13                for _, term := range terms[1:] {
14                    tracer.used(event.Plug(term))
15                }
16            }
17        case *ast.Term:
18            // Standalone expression (3)
19            tracer.used(event.Plug(terms))
20        }
21    }
22}

Terme annotieren

Bei der Implementierung von `used(*ast.Term)` stellt sich die nächste Frage: Wie ordnen wir einen Term einem `Path` in der Eingabe zu?

Eine Möglichkeit wäre, das Eingabedokument nach passenden Termen zu durchsuchen. Das würde aber zu vielen falsch positiven Ergebnissen führen, da eine Zeichenfolge wie `10.0.0.0/24` mehrfach in der Eingabe vorkommen kann!

Stattdessen versehen wir alle Terme mit ihrem Pfad, also wir annotieren sie. Terme in OPA können Metadaten enthalten, darunter die Position in der Rego-Quelldatei. Dieses Feld können wir wiederverwenden, um darin einen Eingabe-`Path` zu speichern. Das ist ein wenig umständlich, aber bei genauerer Betrachtung durchaus vertretbar, da das Feld zum Speichern von Positionen vorgesehen ist.

Der folgende Ausschnitt veranschaulicht, wie wir die ersten Zeilen unserer CloudFormation-Vorlage annotieren möchten:

Resources:                    # ["Resources"]
  Vpc:                        # ["Resources", "Vpc"]
    Type: AWS::EC2::VPC       # ["Resources", "Vpc", "Type"]
    Properties:               # ["Resources", "Vpc", "Properties"]
      CidrBlock: 10.0.0.0/16  # ["Resources", "Vpc", "Properties", "CidrBlock"]

`annotate` durchläuft den Wert rekursiv, um den `Path` jedes Knotens zu bestimmen. Der Einfachheit halber unterstützen wir nur Objekte und lassen Mengen und Arrays außen vor.

1func annotate(path Path, term *ast.Term) {
2    // Annotate current term by setting location.
3    if bytes, err := json.Marshal(path); err == nil {
4        term.Location = &ast.Location{}
5        term.Location.File = "path:" + string(bytes)
6    }
7    // Recursively annotate children.
8    switch value := term.Value.(type) {
9    case ast.Object:
10        for _, key := range value.Keys() {
11            if str, ok := key.Value.(ast.String); ok {
12                path = append(path, string(str))
13                annotate(path, value.Get(key))
14                path = path[:len(path)-1]
15            }
16        }
17    }
18}

Mit dieser Annotation lässt sich `used(*ast.Term)` ganz einfach schreiben. Dabei müssen Sie nur beachten, dass nicht alle Werte annotiert sind. Wir tun dies nur für Werte aus dem Eingabedokument, nicht etwa für Literale, die im Rego-Quellcode enthalten sind.

1func (tracer *locationTracer) used(term *ast.Term) {
2    if term.Location != nil {
3        val := strings.TrimPrefix(term.Location.File, "path:")
4        if len(val) != len(term.Location.File) {
5            // Only when we stripped a "path" suffix.
6            var path Path
7            json.Unmarshal([]byte(val), &path)
8            tracer.tree.Insert(path)
9        }
10    }
11}

Zusammenfassung

Das war’s! Viele Details haben wir ausgelassen, etwa Arrays und die Anwendung auf eine komplexere IaC-Sprache wie HCL.

Zusätzlich markieren wir auch die `Type`-Attribute als verwendet, da wir sie in unserer Policy prüfen. Das ist nicht ideal. Als Alternative versuchen wir, stattdessen eine ressourcenorientierte Rego-API bereitzustellen. Das geht jedoch über den Rahmen dieses Beispiels hinaus.

Wenn Sie mehr über diese Features erfahren möchten, empfehlen wir Ihnen snyk/policy-engine für die Kernimplementierung oder das aktualisierte Snyk IaC. Es bietet dieses und zahlreiche weitere Features, darunter ein umfassendes Regelpaket.

Zum Abschluss folgt eine Hauptfunktion, die alles miteinander verbindet und einige Debugging-Informationen ausgibt. Sie fasst im Wesentlichen die bisher definierten Grundfunktionen zusammen und führt sie anhand eines Beispiels aus. Wir nehmen sie trotzdem auf, damit dieser Beitrag als reproduzierbares, eigenständiges Beispiel funktioniert.

1func infer(policy string, file string) error {
2    source, err := NewSource(file)
3    if err != nil {
4        return err
5    }
6
7    bytes, err := ioutil.ReadFile(file)
8    if err != nil {
9        return err
10    }
11
12    var node yaml.Node
13    if err := yaml.Unmarshal(bytes, &node); err != nil {
14        return err
15    }
16
17    var doc interface{}
18    if err := yaml.Unmarshal(bytes, &doc); err != nil {
19        return err
20    }
21
22    input, err := ast.InterfaceToValue(doc)
23    if err != nil {
24        return err
25    }
26
27    annotate(Path{}, ast.NewTerm(input))
28    if bytes, err = ioutil.ReadFile(policy); err != nil {
29        return err
30    }
31
32    tracer := newLocationTracer()
33    results, err := rego.New(
34        rego.Module(policy, string(bytes)),
35        rego.ParsedInput(input),
36        rego.Query("data.policy.deny"),
37        rego.Tracer(tracer),
38    ).Eval(context.Background())
39    if err != nil {
40        return err
41    }
42
43    fmt.Fprintf(os.Stderr, "Results: %v\n", results)
44    for _, path := range tracer.tree.List() {
45        fmt.Fprintf(os.Stderr, "Location: %s\n", source.Location(path).String())
46    }
47    return nil
48}
49func main() {
50    if err := infer("policy.rego", "template.yml"); err != nil {
51        fmt.Fprintf(os.Stderr, "%s\n", err)
52        os.Exit(1)
53    }
54}

Den vollständigen Code für diesen PoC finden Sie in diesem Gist.