Keto源码解析

源码在https://github.com/ory/keto

以keto serve启动

  • 导入模块初始化

    • 引入viper 初始化环境变量
    • 引入cobra 以便后续注册各个command
    • cmd模块初始化过程中添加serveCmd
  • 在导入server指令时导入核心的几个模块

    • keto/driver
      • 借用viper.AutomaticEnv()环境变量初始化
      • 设置默认监听端口4466
      • 注册两种不同类型的registry: SQL和Memory
      • 导入模块registry
        • 导入模块keto/engine
          • 实现engine 的eval接口
        • 导入模块keto/engine/ladon
          • 实现RBAC模型
          • 实现IAM模型
        • 导入模块keto/storage
          • 实现Memory和SQL两种不同类型StoreManager具体的操作接口
          • 对应handler回调函数实现
    • keto/engine/ladon/rego
      • 编写的rego模板
  • 触发serveCmd的Run指令

    var serveCmd = &cobra.Command{
    	Use:   "serve",
    	Short: "Starts the server and serves the HTTP REST API",
    	Long: `........`,
    	Run: server.RunServe(logger, Version, Commit, Date),
    }
    
  • RunServe流程

    • NewDefaultDriver
      • 使用Viper来初始化配置和环境变量,获得配置c
      • 使用配置c来初始化Registry,根据实际情况启用SQL或者Memory
      • 初始化SQL和DB连接池或者Memory内存池
    • 初始化httprouter
    • ladonEngine绑定router
      • ladon.NewEngine初始化
      • Register
        • 通过kstorage.handler将HTTP请求接口和handler处理函数绑定起来【后面细谈】
    • 初始化中间件系统negroni
    • 初始化监控metrics
    • 初始化cors配置
    • 初始化TLS配置

判断是否允许 XXXX/allowed接口

r.POST(BasePath+"/allowed", e.engine.Evaluate(e.eval))

其中e.engine.Evaluate的实现如下

func (h *Engine) Evaluate(e evaluator) httprouter.Handle {
	return func(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
		ctx := r.Context()

		rs, err := e(ctx, r, ps)
		if err != nil {
			h.h.WriteError(w, r, err)
			return
		}

		allowed, err := h.eval(ctx, rs)
		if err != nil {
			h.h.WriteError(w, r, err)
			return
		}

		code := http.StatusOK
		if !allowed {
			code = http.StatusForbidden
		}

		h.h.WriteCode(w, r, code, &AuthorizationResult{Allowed: allowed})
	}
}
  • 初始化请求上下文
  • evaluator传入的是e.eval,其实现如下,使用该函数编译并解析请求上下文,把编译结果返回
func (e *Engine) eval(ctx context.Context, r *http.Request, ps httprouter.Params) ([]func(*rego.Rego), error) {
	f, err := flavor(ps)
	if err != nil {
		return nil, err
	}

	query := fmt.Sprintf("data.ory.%s.allow", f)
	store, err := e.s.Storage(ctx, schema, []string{policyCollection(f), roleCollection(f)})
	if err != nil {
		return nil, err
	}

	var i Input
	dec := json.NewDecoder(r.Body)
	dec.DisallowUnknownFields()
	if err := dec.Decode(&i); err != nil {
		return nil, errors.WithStack(err)
	}

	return []func(*rego.Rego){
		rego.Query(query),
		rego.Store(store),
		rego.Input(&i),
	}, nil
}

利用flavor函数找到需要的flavor,然后根据映射通过policyCollection和rollCollection拼凑出对应flavor的policy和role path,

func (m *SQLManager) Storage(ctx context.Context, schema string, collections []string) (storage.Store, error) {
	return toRegoStore(ctx, schema, collections, func(i context.Context, s string) ([]json.RawMessage, error) {
		var items []json.RawMessage
		if err := m.db.SelectContext(
			ctx,
			&items,
			m.db.Rebind("SELECT document FROM rego_data WHERE collection=? ORDER BY id ASC"), s,
		); err != nil {
			return nil, errors.WithStack(err)
		}
		return items, nil
	})
}

然后根据path从数据库中找到对应的document,然后解析完成后通过opa的storage写入内存中,并返回store

func toRegoStore(ctx context.Context, schema string, collections []string, query func(context.Context, string) ([]json.RawMessage, error)) (storage.Store, error) {
	var s map[string]interface{}
	dec := json.NewDecoder(bytes.NewBufferString(schema))
	dec.UseNumber()
	if err := dec.Decode(&s); err != nil {
		return nil, errors.WithStack(err)
	}

	db := inmem.NewFromObject(s)
	txn, err := db.NewTransaction(ctx, storage.WriteParams)
	if err != nil {
		return nil, errors.WithStack(err)
	}

	for _, c := range collections {
		path, ok := storage.ParsePath(c)
		if !ok {
			return nil, errors.Errorf("unable to parse storage path: %s", c)
		}

		var val []interface{}
		var b bytes.Buffer

		d, err := query(ctx, c)
		if err != nil {
			return nil, err
		}

		if err := json.NewEncoder(&b).Encode(d); err != nil {
			return nil, errors.WithStack(err)
		}

		dec := json.NewDecoder(&b)
		dec.UseNumber()
		if err := dec.Decode(&val); err != nil {
			return nil, errors.WithStack(err)
		}

		if err := db.Write(ctx, txn, storage.AddOp, path, val); err != nil {
			return nil, errors.WithStack(err)
		}
	}

	if err := db.Commit(ctx, txn); err != nil {
		return nil, errors.WithStack(err)
	}
	return db, nil
}

然后通过读取request的body获取input数据

  • 后面通过执行h.eval(ctx, rs) 来判断是否允许该操作
func (h *Engine) eval(ctx context.Context, options []func(*rego.Rego)) (bool, error) {
	//tracer := topdown.NewBufferTracer()
	r := rego.New(
		append(
			options,
			rego.Compiler(h.compiler),
			//rego.Tracer(tracer),
		)...,
	)

	rs, err := r.Eval(ctx)
	if err != nil {
		return false, errors.WithStack(err)
	}

	if len(rs) != 1 || len(rs[0].Expressions) != 1 {
		return false, errors.Errorf("expected one evaluation result but got %d results instead", len(rs))
	}

	result, ok := rs[0].Expressions[0].Value.(bool)
	if !ok {
		return false, errors.Errorf("expected evaluation result to be of type bool but got %T instead", rs[0].Expressions[0].Value)
	}

	return result, nil
}