pinned-package-updater/internal/parser/debian.go

/*
Copyright © 2021 Drew Short <warrick@sothr.com>

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package parser

import (
	"bytes"
	"errors"
	"strings"

	log "github.com/sirupsen/logrus"

	"mvdan.cc/sh/v3/syntax"
)

// DebianCommandParser the parser responsible for handling Debian based images
type DebianCommandParser struct {
	parser  *syntax.Parser
	printer *syntax.Printer
}

// newDebianCommandParser Provision a new instance of DebianCommandParser with required components
func newDebianCommandParser() DebianCommandParser {
	return DebianCommandParser{
		parser:  syntax.NewParser(),
		printer: syntax.NewPrinter(),
	}
}

// shellNodeToString Convert a node to the string representation
func (c DebianCommandParser) shellNodeToString(node syntax.Node) string {
	buffer := bytes.NewBufferString("")
	err := c.printer.Print(buffer, node)
	if err != nil {
		log.Panic("Failed to convert node to string")
	}
	return buffer.String()
}

// isRedirectedToAptSourcesList Checks if the list of redirects contains the apt sources list directory
func isRedirectedToAptSourcesList(redirects []*syntax.Redirect) bool {
	if redirects != nil && len(redirects) > 0 {
		for _, redirect := range redirects {
			if strings.Contains(redirect.Word.Lit(), "/etc/apt/sources.list") {
				return true
			}
		}
	}
	return false
}

// isAddRepositoryExpression Checks for expressions that add to the repository
func isAddRepositoryExpression(command syntax.CallExpr, redirects []*syntax.Redirect) bool {
	// Adding via debian tooling
	if command.Args[0].Lit() == "add-apt-repository" {
		return true
		// Echoing into a sources file directly
	} else if command.Args[0].Lit() == "echo" && isRedirectedToAptSourcesList(redirects) {
		return true
	}
	return false
}

// trimQuotes return a string without surrounding quotes
func trimQuotes(value string) string {
	if value[0] == '"' && value[len(value)-1] == '"' {
		return value[1 : len(value)-1]
	} else if value[0] == '\'' && value[len(value)-1] == '\'' {
		return value[1 : len(value)-1]
	}
	return value
}

// getAddedRepository parse the repository information from the shell commands
func (c DebianCommandParser) getAddedRepository(command syntax.CallExpr, redirects []*syntax.Redirect) (string, error) {
	// Adding via debian tooling
	if command.Args[0].Lit() == "add-apt-repository" {
		return command.Args[len(command.Args)-1].Lit(), nil
		// Echoing into a sources file directly
	} else if command.Args[0].Lit() == "echo" && isRedirectedToAptSourcesList(redirects) {
		for _, argument := range command.Args {
			var argumentNode syntax.Node = argument
			argumentValue := trimQuotes(c.shellNodeToString(argumentNode))
			if strings.Contains(argumentValue, "deb") || strings.Contains(argumentValue, "http://") {
				return argumentValue, nil
			}
		}
	}
	return "", errors.New("failed to parse repository")
}

// GetRepositories Returns a list of Repository entries that were identified
func (c DebianCommandParser) GetRepositories(command string) ([]*Repository, error) {
	var repositories []*Repository

	reader := strings.NewReader(command)
	parsedScript, err := c.parser.Parse(reader, "")
	if err != nil {
		log.Fatalf("Failed to parse script \"%s\"", command)
		return nil, err
	}

	syntax.Walk(parsedScript, func(node syntax.Node) bool {
		switch parsedStatement := node.(type) {
		case *syntax.Stmt:
			switch parsedCommand := parsedStatement.Cmd.(type) {
			case *syntax.CallExpr:
				if isAddRepositoryExpression(*parsedCommand, parsedStatement.Redirs) {
					repositoryInformation, err := c.getAddedRepository(*parsedCommand, parsedStatement.Redirs)
					if err != nil {
						log.Fatalf("Unexpected error while parsing repository from \"%s\": %s", c.shellNodeToString(node), err)
					}
					repositories = append(repositories, &Repository{Information: repositoryInformation})
					log.Debugf("Found repository \"%s\"", repositoryInformation)
				}
			}
		}
		return true
	})

	return repositories, nil
}

// isAddRepositoryExpression Checks for expressions that add to the repository
func isInstallerExpression(command syntax.CallExpr) bool {
	// Adding via debian tooling
	commandProgram := command.Args[0].Lit()
	if (commandProgram == "apt-get" || commandProgram == "apt") && len(command.Args) > 1 && command.Args[1].Lit() == "install" {
		return true
	}
	return false
}

func (c DebianCommandParser) getInstalledPinnedPackages(command syntax.CallExpr) ([]*Package, error) {
	var packages []*Package

	// Adding via debian tooling
	commandProgram := command.Args[0].Lit()
	if (commandProgram == "apt-get" || commandProgram == "apt") && len(command.Args) > 1 && command.Args[1].Lit() == "install" {
		for _, argument := range command.Args {
			var argumentNode syntax.Node = argument
			argumentValue := trimQuotes(c.shellNodeToString(argumentNode))
			if strings.Contains(argumentValue, "=") {
				pinnedPackageParts := strings.SplitN(argumentValue, "=", 2)
				packages = append(packages, &Package{
					Name:    pinnedPackageParts[0],
					Version: pinnedPackageParts[1],
				})
			}
		}
		return packages, nil
	}
	return packages, errors.New("failed to parse packages")
}

// GetPinnedPackages get a list of all pinned packages
func (c DebianCommandParser) GetPinnedPackages(command string) ([]*Package, error) {
	var packages []*Package

	reader := strings.NewReader(command)
	parsedScript, err := c.parser.Parse(reader, "")
	if err != nil {
		log.Fatalf("Failed to parse script \"%s\"", command)
		return nil, err
	}

	syntax.Walk(parsedScript, func(node syntax.Node) bool {
		switch parsedStatement := node.(type) {
		case *syntax.Stmt:
			switch parsedCommand := parsedStatement.Cmd.(type) {
			case *syntax.CallExpr:
				if isInstallerExpression(*parsedCommand) {
					pinnedPackages, err := c.getInstalledPinnedPackages(*parsedCommand)
					if err != nil {
						log.Fatalf("Unexpected error while parsing pinned packages from \"%s\": %s", c.shellNodeToString(node), err)
					}
					for _, pinnedPackage := range pinnedPackages {
						packages = append(packages, pinnedPackage)
						log.Debugf("Found pinned package \"%s=%s\"", pinnedPackage.Name, pinnedPackage.Version)
					}
				}
			}
		}
		return true
	})

	return packages, nil
}