// Copyright 2017 Istio Authors
//
// 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 kube

import (
	"reflect"
	"time"

	"istio.io/istio/pkg/spiffe"

	v1 "k8s.io/api/core/v1"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/watch"
	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
	"k8s.io/client-go/tools/cache"

	"istio.io/istio/security/pkg/listwatch"
	"istio.io/istio/security/pkg/registry"
	"istio.io/pkg/log"
)

// ServiceAccountController monitors service account definition changes in a namespace.
// For each service account object, its SpiffeID is added to identity registry for
// whitelisting purpose.
type ServiceAccountController struct {
	core corev1.CoreV1Interface

	// identity registry object
	reg registry.Registry

	// controller for service objects
	controller cache.Controller
}

// NewServiceAccountController returns a new ServiceAccountController
func NewServiceAccountController(core corev1.CoreV1Interface, namespaces []string, reg registry.Registry) *ServiceAccountController {
	c := &ServiceAccountController{
		core: core,
		reg:  reg,
	}

	LW := listwatch.MultiNamespaceListerWatcher(namespaces, func(namespace string) cache.ListerWatcher {
		return &cache.ListWatch{
			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
				return core.ServiceAccounts(namespace).List(options)
			},
			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
				return core.ServiceAccounts(namespace).Watch(options)
			},
		}
	})

	_, c.controller = cache.NewInformer(LW, &v1.ServiceAccount{}, time.Minute, cache.ResourceEventHandlerFuncs{
		AddFunc:    c.serviceAccountAdded,
		DeleteFunc: c.serviceAccountDeleted,
		UpdateFunc: c.serviceAccountUpdated,
	})
	return c
}

// Run starts the ServiceAccountController until a value is sent to stopCh.
// It should only be called once.
func (c *ServiceAccountController) Run(stopCh chan struct{}) {
	go c.controller.Run(stopCh)
}

func getSpiffeID(sa *v1.ServiceAccount) string {
	return spiffe.MustGenSpiffeURI(sa.GetNamespace(), sa.GetName())
}

func (c *ServiceAccountController) serviceAccountAdded(obj interface{}) {
	sa := obj.(*v1.ServiceAccount)
	id := getSpiffeID(sa)
	err := c.reg.AddMapping(id, id)
	if err != nil {
		log.Errorf("cannot add mapping %q -> %q to registry: %s", id, id, err.Error())
	}
}

func (c *ServiceAccountController) serviceAccountDeleted(obj interface{}) {
	sa, ok := obj.(*v1.ServiceAccount)
	if !ok {
		// OnDelete can get an object of type DeletedFinalStateUnknown if it misses the delete event.
		// Citadel should not proceed in that case
		log.Warnf("Failed to convert to serviceaccount object: %v", obj)
		return
	}
	id := getSpiffeID(sa)
	err := c.reg.DeleteMapping(id, id)
	if err != nil {
		log.Errorf("cannot delete mapping %q to %q from registry: %s", id, id, err.Error())
	}

}

func (c *ServiceAccountController) serviceAccountUpdated(oldObj, newObj interface{}) {
	if oldObj == newObj || reflect.DeepEqual(oldObj, newObj) {
		// Nothing is changed. The method is invoked by periodical re-sync with the apiserver.
		return
	}

	oldSa := oldObj.(*v1.ServiceAccount)
	newSa := newObj.(*v1.ServiceAccount)
	// if name or namespace has changed
	if oldSa.GetName() != newSa.GetName() || oldSa.GetNamespace() != newSa.GetNamespace() {
		oldID := getSpiffeID(oldSa)
		newID := getSpiffeID(newSa)
		_ = c.reg.DeleteMapping(oldID, oldID)
		_ = c.reg.AddMapping(newID, newID)
	}
}
