Ubicaciones automáticas del código fuente con Rego

En Snyk, somos grandes admiradores de Rego de Open Policy Agent. Snyk IaC se basa en un amplio conjunto de reglas escritas en Rego, y los clientes también pueden agregar sus propias reglas personalizadas.

Hace poco lanzamos una serie de mejoras para Snyk IaC y, en esta publicación del blog, analizamos en profundidad una función particularmente interesante: las ubicaciones automáticas del código fuente para las infracciones de reglas.

Al comprobar los archivos de IaC en busca de problemas conocidos, el comando actualizado `snyk iac test` mostrará información precisa sobre el archivo, la línea y la columna de cada infracción de regla. Esto también funciona con reglas personalizadas, sin que el usuario tenga que hacer nada.

Pero antes de presentar una prueba de concepto independiente para esta técnica, tendremos que hacer algunas simplificaciones. La implementación completa está disponible en nuestro motor de políticas unificado.

Empecemos con un ejemplo de CloudFormation. Aunque nuestro motor de IaC admite muchos formatos y se enfoca especialmente en Terraform, CloudFormation es un buen ejemplo porque podemos analizarlo sin demasiadas dependencias (después de todo, es solo 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

Queremos asegurarnos de que ninguna subred use un bloque CIDR mayor que `/24`, así que escribamos una política de Rego para hacerlo:

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}

De esta manera, `deny` generará un conjunto de recursos denegados. No explicaremos en detalle cómo funciona Rego, pero si quieres obtener más información, te recomendamos el excelente curso OPA by Example.

Podemos dividir el problema en dos partes:

1. Queremos inferir que nuestra política usa el atributo `CidrBlock`

2. Luego, recuperaremos la ubicación en el código fuente

Empecemos por (2), ya que nos permite familiarizarnos con el código.

Recuperación de la ubicación en el código fuente

Una ubicación en el código fuente se ve así:

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}

También presentaremos un tipo auxiliar para representar rutas en YAML. En YAML, hay dos tipos de documentos anidados: arreglos y objetos.

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

Si quisiéramos poder hacer referencia a cualquier subdocumento, podríamos usar algo similar a las rutas JSON. En el ejemplo anterior, [`"some_array", 1`] apuntaría a `"word"`. Pero como no admitiremos arreglos en nuestra prueba de concepto, podemos conformarnos con usar un arreglo de cadenas.

type Path []string

Un ejemplo de una ruta sería algo como:

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

Ahora podemos ofrecer un tipo práctico para cargar YAML e indicarnos la `Location` de ciertas `Paths`.

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}

Para encontrar la ubicación en el código fuente de una `Path`, hay que recorrer un árbol de nodos YAML:

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}

Conjuntos y árboles de rutas

Con esto resuelto, redujimos el problema: en lugar de inferir automáticamente las ubicaciones en el código fuente que se usan en una política, inferiremos automáticamente las rutas de atributos.

Esto también es importante por otros motivos. Por ejemplo, Snyk puede aplicar las mismas políticas tanto a los recursos de IaC como a los recursos detectados mediante análisis de la nube. Estos últimos no tienen ubicaciones significativas en el código fuente, pero sí cuentan con rutas de atributos útiles.

Por lo tanto, queremos definir conjuntos de rutas de atributos. Como las rutas se respaldan con arreglos, lamentablemente no podemos usar algo como `map[Path]struct{}` como conjunto en Go.

En su lugar, tendremos que almacenarlas en un árbol recursivo.

type PathTree map[string]PathTree

Esta representación tiene otras ventajas. En general, solo nos interesan las rutas más largas que usa una política, ya que son más específicas. Nuestra política de ejemplo usa `Path{"Resources", "PrivateSubnet", "Properties"}` y también `Path{"Resources", "PrivateSubnet", "Properties", "CidrBlock"}`; solo nos interesa esta última.

Definiremos un método recursivo para insertar una `Path` en nuestro árbol:

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}

…y también una manera de obtener una lista de `Paths`. Esto genera algunas asignaciones de memoria innecesarias, pero podemos aceptarlo.

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}

Ahora tenemos una forma práctica de almacenar las `Paths` que usó una política, y podemos convertirlas en ubicaciones del código fuente.

Análisis estático frente a análisis en tiempo de ejecución

La siguiente pregunta es cómo determinar qué `Paths` de una entrada determinada usa una política y luego insertarlas en el árbol.

No es una pregunta sencilla, ya que el código puede manipular la entrada de distintas maneras antes de usar las rutas. Tendremos que revisar tanto las funciones definidas por el usuario ( `has_bad_subnet`) como las funciones integradas ( `object.get`), solo para ilustrar uno de los posibles obstáculos:

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}

Por suerte, no somos los primeros en preguntarnos qué hacen los programas: la gente siente curiosidad por esto desde que se escribió el primero. Por lo general, hay dos maneras de responder una pregunta como esta sobre un fragmento de código:

• Análisis estático: Intentar responder examinando el árbol de sintaxis, los tipos y otra información estática que podamos obtener del intérprete de OPA (o agregarle). La ventaja es que no necesitamos ejecutar esta política, lo cual es ideal si no confiamos en quienes la escribieron. La desventaja es que las técnicas de análisis estático suelen generar falsos negativos y falsos positivos.

• Análisis en tiempo de ejecución: Rastrear la ejecución de políticas específicas e inferir qué `Paths` se usan a partir de la información en tiempo de ejecución. La desventaja es que tenemos que ejecutar la política, y agregar este análisis puede ralentizar su evaluación.

Probamos ambos enfoques, pero decidimos optar por el segundo porque nos resultó mucho más fácil implementarlo de forma confiable y el impacto en el rendimiento fue insignificante. También vale la pena mencionar que no se trata de una elección excluyente: podrías adoptar un enfoque híbrido y combinar ambos.

OPA ofrece una interfaz Tracer que permite recibir eventos sobre lo que hace el intérprete. Un caso de uso común de los rastreadores es enviar métricas o información de depuración a un registro centralizado. Sin embargo, hoy la usaremos para otra cosa.

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}

Rastrear el uso de términos

Rego es un lenguaje expresivo. Aunque se realizan algunas transformaciones para simplificarlo y reducirlo a un formato más simple para el intérprete, sigue habiendo una cantidad considerable de eventos.

Solo nos interesan dos de ellos. Consideramos que un valor se usa si:

1. Se unifica (puedes pensar en esto como una asignación; no entraremos en detalles) con otra expresión, como:

x = input.Foo

Esto también incluye `==` y `:=`. Como esta prueba puede fallar, podemos afirmar que usamos tanto el lado izquierdo como el derecho.

2. Se usa como argumento de una función integrada, como:

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

Aunque Rego toma algunos conceptos de los lenguajes perezosos, los argumentos de las funciones integradas siempre están completamente evaluados antes de invocar la función. Por lo tanto, podemos decir que usamos todos los argumentos que se pasan a la función integrada.

3. Se usa como expresión independiente, por ejemplo:

volume.encrypted

Esto se usa comúnmente para evaluar booleanos y comprobar que existan atributos.

Ahora es momento de implementar. Buscamos dos eventos y delegamos el trabajo a una función específica para que el código sea un poco más legible:

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}

Más adelante nos encargaremos de insertar elementos en nuestro `PathTree` mediante una función auxiliar llamada `used(*ast.Term)`. Por ahora, marquemos como usados tanto el lado izquierdo como el derecho de la unificación:

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` es una función auxiliar que reemplaza las variables por sus valores reales.

Un evento `EvalOp` cubre los casos (2) y (3) mencionados antes. En el caso de una función integrada, tendremos un arreglo de términos: el primer elemento es la función y los demás son los argumentos. Podemos comprobar que se trata de una función integrada consultando `ast.BuiltinMap`.

El caso de una expresión independiente es sencillo.

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}

Anotar términos

Al intentar implementar `used(*ast.Term)`, surge la siguiente pregunta: dado un término, ¿cómo lo vinculamos con una `Path` de la entrada?

Una opción sería buscar términos coincidentes en el documento de entrada. Pero eso produciría muchos falsos positivos, ya que una cadena como `10.0.0.0/24` podría aparecer varias veces en la entrada.

En su lugar, vamos a anotar todos los términos con su ruta. Los términos en OPA pueden incluir metadatos, como la ubicación en el archivo fuente de Rego. Podemos reutilizar este campo para almacenar una `Path` de entrada. Es un poco improvisado, pero, si lo miramos con buenos ojos, no nos estamos desviando demasiado: el campo está pensado para almacenar ubicaciones.

El siguiente fragmento muestra cómo queremos anotar las primeras líneas de nuestra plantilla de CloudFormation:

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` implementa un recorrido recursivo para determinar la `Path` de cada nodo del valor. Para mantener la concisión, solo admitimos objetos y dejamos fuera los conjuntos y los arreglos.

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}

Con esta anotación, es fácil escribir `used(*ast.Term)`. Lo único que hay que tener en cuenta es que no todos los valores están anotados. Solo lo hacemos con los que provienen del documento de entrada, no con, por ejemplo, los literales incluidos en el código fuente de Rego.

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}

Conclusión

¡Eso es todo! Omitimos muchos detalles, como el uso de arreglos y cómo aplicar esto a un lenguaje de IaC más complejo, como HCL.

Además, también marcamos los atributos `Type` como usados, ya que los comprobamos en nuestra política. Esto no es lo ideal y, como alternativa, intentamos ofrecer una API de Rego orientada a los recursos. Pero eso queda fuera del alcance de este ejemplo.

Si te interesa obtener más información sobre cualquiera de estas funciones, te recomendamos consultar snyk/policy-engine para conocer la implementación principal o la versión actualizada de Snyk IaC, que incluye esta función y muchas otras, entre ellas un paquete exhaustivo de reglas.

A continuación, presentamos una función principal que integra todo y muestra información de depuración. En su mayor parte, solo reúne las primitivas que definimos hasta ahora y las ejecuta con un ejemplo. Pero la incluimos para que esta publicación sirva como ejemplo independiente y reproducible.

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}

Puedes encontrar el código completo de esta prueba de concepto en este gist.