Localização automática do código-fonte com Rego

Na Snyk, somos grandes fãs do Rego do Open Policy Agent. O Snyk IaC é baseado em um amplo conjunto de regras escritas em Rego, e os clientes também podem adicionar suas próprias regras personalizadas.

Lançamos recentemente uma série de melhorias no Snyk IaC. Neste post, vamos nos aprofundar em um recurso especialmente interessante: a localização automática do código-fonte para violações de regras.

Ao verificar arquivos IaC em busca de problemas conhecidos, o comando atualizado `snyk iac test` exibe informações precisas de arquivo, linha e coluna para cada violação de regra. Isso também funciona com regras personalizadas, sem exigir nenhum trabalho de quem usa o comando.

Mas, antes de apresentarmos uma prova de conceito independente dessa técnica, precisamos fazer algumas simplificações. A implementação completa está disponível no nosso mecanismo de políticas unificado.

Vamos começar com um exemplo de CloudFormation. Embora nosso mecanismo IaC ofereça suporte a vários formatos, com foco especial em Terraform, o CloudFormation é um bom exemplo, pois podemos analisá-lo sem muitas dependências (afinal, é apenas 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 garantir que nenhuma sub-rede use um bloco CIDR maior que `/24`. Para isso, vamos escrever uma política Rego:

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}

Assim, `deny` produzirá um conjunto de recursos negados. Não vamos explicar em detalhes como Rego funciona, mas, se quiser saber mais, recomendamos o excelente curso OPA by Example.

Podemos dividir o problema em duas partes:

1. Precisamos inferir que nossa política usa o atributo `CidrBlock`

2. Em seguida, vamos recuperar a localização no código-fonte

Vamos começar pela parte (2), que é uma boa maneira de nos familiarizarmos com o código.

Recuperação da localização no código-fonte

Uma localização no código-fonte é assim:

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}

Também vamos introduzir um tipo auxiliar para representar caminhos em YAML. No YAML, há dois tipos de documentos aninhados: arrays e objetos.

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

Para poder fazer referência a qualquer subdocumento, poderíamos usar algo parecido com caminhos JSON. No exemplo acima, [`"some_array", 1`] apontaria para `"word"`. Mas, como não vamos oferecer suporte a arrays nesta prova de conceito, podemos usar apenas um array de strings.

type Path []string

Um exemplo de caminho seria algo assim:

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

Agora podemos criar um tipo auxiliar para carregar YAML e informar a `Location` de determinados `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 a localização no código-fonte de um `Path`, basta percorrer uma árvore de nós 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 e árvores de caminhos

Com isso, reduzimos o problema: em vez de inferir automaticamente as localizações no código-fonte usadas por uma política, vamos inferir automaticamente os caminhos de atributos.

Isso também é importante por outros motivos. Por exemplo, a Snyk pode aplicar as mesmas políticas tanto a recursos IaC quanto a recursos descobertos por meio de varreduras na nuvem. Estes últimos não têm localizações no código-fonte realmente significativas, mas têm caminhos de atributos relevantes!

Portanto, queremos definir conjuntos de caminhos de atributos. Como os caminhos são representados por arrays, infelizmente não podemos usar algo como `map[Path]struct{}` para criar um conjunto em Go.

Em vez disso, precisamos armazená-los em uma árvore recursiva.

type PathTree map[string]PathTree

Essa representação também tem outras vantagens. Em geral, só nos interessam os caminhos mais longos usados por uma política, pois são mais específicos. Nossa política de exemplo usa `Path{"Resources", "PrivateSubnet", "Properties"}` e também `Path{"Resources", "PrivateSubnet", "Properties", "CidrBlock"}` — só nos interessa o último.

Vamos definir um método recursivo para inserir um `Path` na nossa árvore:

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}

…e também uma maneira de obter uma lista de `Paths`. Isso gera algumas alocações desnecessárias, mas podemos conviver com isso.

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}

Agora temos uma maneira organizada de armazenar os `Paths` usados por uma política e de convertê-los em localizações no código-fonte.

Análise estática vs. análise em tempo de execução

A próxima questão é descobrir quais `Paths` de uma determinada entrada são usados por uma política e, em seguida, inseri-los na árvore com `Insert`.

Essa não é uma questão simples, pois o código pode manipular a entrada de diferentes maneiras antes de usar os caminhos. Precisamos analisar funções definidas pelo usuário ( `has_bad_subnet`) e também funções integradas ( `object.get`), para ilustrar apenas um dos possíveis 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}

Felizmente, não estamos sozinhos nessa questão: desde que o primeiro programa foi escrito, as pessoas têm curiosidade sobre o que os programas fazem. Em geral, há duas maneiras de responder a uma pergunta desse tipo sobre um trecho de código:

• Análise estática: tentar responder examinando a árvore sintática, os tipos e outras informações estáticas que podemos obter (ou adicionar) ao interpretador OPA. A vantagem é que não precisamos executar essa política, o que é ótimo se não confiamos em quem a criou. A desvantagem é que as técnicas de análise estática costumam gerar falsos negativos e falsos positivos.

• Análise em tempo de execução: rastrear a execução de políticas específicas e inferir quais `Paths` estão sendo usados com base nas informações de execução. A desvantagem é que precisamos executar a política, e essa análise pode deixar a avaliação da política mais lenta.

Testamos as duas abordagens, mas escolhemos a última porque descobrimos que era muito mais fácil implementá-la com confiabilidade, e o impacto no desempenho era insignificante. Vale lembrar também que não é uma escolha entre uma coisa ou outra: podemos adotar uma abordagem híbrida e combinar as duas.

O OPA oferece uma interface Tracer que pode ser usada para receber eventos sobre o que o interpretador está fazendo. Um uso comum dos rastreadores é enviar métricas ou informações de depuração para algum log centralizado. Hoje, porém, vamos usá-la para outra finalidade.

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}

Rastreamento do uso de termos

Rego é uma linguagem expressiva. Mesmo que algumas transformações simplifiquem o código para o interpretador, ainda há uma quantidade considerável de eventos.

Estamos interessados em apenas dois deles. Consideramos que um valor é usado se:

1. Ele é unificado (pense nisso como uma atribuição; não vamos entrar em detalhes) com outra expressão, por exemplo:

x = input.Foo

Isso também abrange `==` e `:=`. Como esse teste pode falhar, podemos dizer que usamos tanto o lado esquerdo quanto o lado direito.

2. Ele é usado como argumento de uma função integrada, por exemplo:

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

Embora Rego aproveite alguns conceitos das linguagens lazy, os argumentos das funções integradas sempre são totalmente avaliados antes que a função seja chamada. Portanto, podemos dizer que usamos todos os argumentos fornecidos à função integrada.

3. Ele é usado como uma expressão independente, por exemplo:

volume.encrypted

Isso é usado com frequência para avaliar valores booleanos e verificar se os atributos existem.

Agora é hora de implementar. Vamos tratar de dois eventos e delegar a uma função específica para deixar o código um pouco mais legível:

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}

Vamos tratar da inserção na nossa `PathTree` mais adiante, em uma função auxiliar chamada `used(*ast.Term)`. Por enquanto, vamos marcar como usados os dois lados da unificação:

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` é uma função auxiliar que preenche as variáveis com seus valores reais.

Um evento `EvalOp` abrange os casos (2) e (3) mencionados acima. No caso de uma função integrada, teremos um array de termos: o primeiro elemento é a função, e os demais são os argumentos. Podemos verificar se estamos lidando com uma função integrada consultando `ast.BuiltinMap`.

O caso de uma expressão independente é simples.

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}

Anotação de termos

Ao tentar implementar `used(*ast.Term)`, surge a próxima questão: dado um termo, como associá-lo a um `Path` na entrada?

Uma opção seria procurar termos correspondentes no documento de entrada. Mas isso geraria muitos falsos positivos, pois uma string como `10.0.0.0/24` pode aparecer várias vezes na entrada!

Em vez disso, vamos anotar todos os termos com o caminho correspondente. Os termos no OPA podem conter alguns metadados, inclusive a localização no arquivo-fonte do Rego. Podemos reutilizar esse campo para armazenar um `Path` de entrada. É um pouco improvisado, mas, pensando bem, estamos usando o campo de uma forma perfeitamente adequada, já que ele foi feito para armazenar localizações.

O trecho a seguir mostra como queremos anotar as primeiras linhas do nosso template do 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 uma travessia recursiva para determinar o `Path` em cada nó do valor. Para manter o exemplo conciso, só oferecemos suporte a objetos, deixando conjuntos e arrays de fora.

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}

Com essa anotação, é fácil escrever `used(*ast.Term)`. Só precisamos lembrar que nem todos os valores estão anotados. Fazemos isso apenas com os valores que vêm do documento de entrada, não, por exemplo, com literais incorporados ao código-fonte 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}

Conclusão

É isso, pessoal! Deixamos de lado muitos detalhes, como arrays e a aplicação dessa técnica a uma linguagem IaC mais complexa, como HCL.

Além disso, também estamos marcando os atributos `Type` como usados, já que os verificamos na nossa política. Isso não é ideal e, como alternativa, tentamos oferecer uma API Rego orientada a recursos. Mas isso foge do escopo deste exemplo.

Se quiser saber mais sobre qualquer um desses recursos, recomendamos conferir snyk/policy-engine para ver a implementação principal ou o Snyk IaC atualizado, que inclui esse recurso e muitos outros, como um conjunto abrangente de regras.

A seguir, temos uma função principal que reúne tudo e exibe algumas informações de depuração. Ela basicamente combina os elementos que definimos até aqui e os executa em um exemplo. Mas vamos incluí-la para que este post funcione como um exemplo independente e reproduzível.

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}

O código completo desta prova de conceito está disponível neste gist.