github_mirrors/attachment
3
0
Fork 1
mirror of https://github.com/openappsec/attachment.git synced 2025-11-20 19:06:41 +03:00
Code Issues Packages Projects Releases Wiki Activity

pulling from dev

Browse Source
This commit is contained in:
wiaamm
2025-01-05 12:01:24 +02:00
parent cfeae10dcf
commit f6d7d09bae
26 changed files with 657 additions and 426 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
attachments/envoy/build_template Normal file → Executable file
View File

@@ -10,4 +10,4 @@ ENVOY_ATTACHMENT_DIR="@ENVOY_ATTACHMENT_DIR@"
cd $ENVOY_ATTACHMENT_DIR
# Run the go build command
CC=gcc CGO_CFLAGS="-I@ATTACHMENTS_INCLUDE_DIR@ -I@NANO_ATTACHMENT_INCLUDE_DIR@" go build -o ${ENVOY_ATTACHMENT_DIR}/libenvoy_attachment.so -buildmode=c-shared -ldflags="-extldflags '-L${SHMEM_LIBRARY_DIR} -L${NANO_ATTACHMENT_LIBRARY_DIR} -L${NANO_ATTACHMENT_UTIL_LIBRARY_DIR} ${LIBRARIES}'"
CGO_CFLAGS="-I@ATTACHMENTS_INCLUDE_DIR@ -I@NANO_ATTACHMENT_INCLUDE_DIR@" go build -o ${ENVOY_ATTACHMENT_DIR}/libenvoy_attachment.so -buildmode=c-shared -ldflags="-extldflags '-L${SHMEM_LIBRARY_DIR} -L${NANO_ATTACHMENT_LIBRARY_DIR} -L${NANO_ATTACHMENT_UTIL_LIBRARY_DIR} ${LIBRARIES}'"

194
attachments/envoy/config.go
View File

@@ -3,7 +3,6 @@ package main
import (
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"runtime"
"strconv"
@@ -35,53 +34,58 @@ unsigned long get_thread_id() {
import "C"
const Name = "cp_nano_filter"
const Admin_api = "http://127.0.0.1:%s/server_info"
const admin_api_server_info = "http://127.0.0.1:%s/server_info"
const keep_alive_interval = 10 * time.Second
var filter_id atomic.Int64
type nano_attachment C.struct_NanoAttachment
var attachments_map map[int]*nano_attachment = nil
var thread_to_attachment_mapping map[int]int = nil
var attachment_to_thread_mapping map[int]int = nil
var attachment_to_filter_request_structs map[int]*filterRequestStructs = nil
var mutex sync.Mutex
const keep_alive_interval = 10 * time.Second
var last_keep_alive time.Time
type nano_attachment C.struct_NanoAttachment
// EnvoyServerInfo represents the structure of the JSON response from /server_info
type EnvoyServerInfo struct {
Concurrency int `json:"concurrency"`
}
// getEnvoyConcurrency fetches and returns the concurrency level of Envoy from the admin API
func getEnvoyConcurrency(admin_api_address string) (int, error) {
resp, err := http.Get(admin_api_address)
if err != nil {
return 0, fmt.Errorf("failed to reach Envoy admin API: %w", err)
}
defer resp.Body.Close()
func getEnvoyConcurrency() int {
concurrency_method := getEnv("CONCURRENCY_CALC", "numOfCores")
if resp.StatusCode != http.StatusOK {
return 0, fmt.Errorf("unexpected status code from Envoy admin API: %d", resp.StatusCode)
if concurrency_method == "numOfCores" {
api.LogWarnf("using number of CPU cores")
return runtime.NumCPU()
}
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
return 0, fmt.Errorf("failed to read response body: %w", err)
var conc_number string
switch concurrency_method {
case "istioCpuLimit":
conc_number = getEnv("ISTIO_CPU_LIMIT", "-1")
api.LogWarnf("using istioCpuLimit, conc_number %s", conc_number)
case "custom":
conc_number = getEnv("CONCURRENCY_NUMBER", "-1")
api.LogWarnf("using custom concurrency number, conc_number %s", conc_number)
default:
api.LogWarnf("unknown concurrency method %s, using number of CPU cores", concurrency_method)
return runtime.NumCPU()
}
var info EnvoyServerInfo
if err := json.Unmarshal(body, &info); err != nil {
return 0, fmt.Errorf("failed to parse JSON response: %w", err)
if conc_number == "-1" {
api.LogWarnf("concurrency number is not set as an env variable, using number of CPU cores")
return runtime.NumCPU()
}
return info.Concurrency, nil
conc_num, err := strconv.Atoi(conc_number)
if err != nil || conc_num <= 0 {
api.LogWarnf("error converting concurrency number %s, using number of CPU cores", conc_number)
return runtime.NumCPU()
}
return conc_num
}
func configurationServer() {
@@ -150,8 +154,7 @@ func configurationServer() {
func init() {
last_keep_alive = time.Time{}
envoyHttp.RegisterHttpFilterFactoryAndConfigParser(Name, ConfigFactory, &parser{})
//envoyHttp.RegisterHttpFilterConfigFactoryAndParser(Name, ConfigFactory, &parser{})
envoyHttp.RegisterHttpFilterConfigFactoryAndParser(Name, ConfigFactory, &parser{})
go configurationServer()
}
@@ -193,37 +196,7 @@ func (p *parser) Parse(any *anypb.Any, callbacks api.ConfigCallbackHandler) (int
return conf, nil
}
var num_of_workers int
concurrency_method := getEnv("CONCURRENCY_CALC", "numOfCores")
if concurrency_method == "numOfCores" {
num_of_workers = runtime.NumCPU()
api.LogInfof("using number of cpu cores %d", num_of_workers)
} else if concurrency_method == "config" {
config_port := getEnv("CONFIG_PORT", "15000")
admin_api := fmt.Sprintf(Admin_api, config_port)
workers, err := getEnvoyConcurrency(admin_api)
if err != nil {
api.LogWarnf("unable to fetch concurrency from admin server, using cpu cores. err: %s", err.Error())
num_of_workers = runtime.NumCPU()
} else {
num_of_workers = workers
}
} else if concurrency_method == "custom" {
conc_number := getEnv("CONCURRENCY_NUMBER", "-1")
if conc_number == "-1" {
api.LogWarnf("concurrency number is not set as an env variable, using cpu cores")
num_of_workers = runtime.NumCPU()
} else if conc_num, err := strconv.Atoi(conc_number); err == nil && conc_num > 0 {
num_of_workers = conc_num
} else {
api.LogWarnf("error converting conc_number %s, using num of cpu cores", conc_number)
num_of_workers = runtime.NumCPU()
}
} else {
api.LogWarnf("unable to fetch concurrency from %s, using cpu cores", concurrency_method)
num_of_workers = runtime.NumCPU()
}
num_of_workers := getEnvoyConcurrency()
configStruct := &xds.TypedStruct{}
if err := any.UnmarshalTo(configStruct); err != nil {
@@ -266,87 +239,46 @@ func (p *parser) Merge(parent interface{}, child interface{}) interface{} {
return &newConfig
}
// func ConfigFactory(c interface{}) api.StreamFilterFactory {
// conf, ok := c.(*config)
// if !ok {
// panic("unexpected config type")
// }
// return func(callbacks api.FilterCallbackHandler) api.StreamFilter {
// worker_thread_id := int(C.get_thread_id())
// api.LogDebugf("worker_thread_id: %d", worker_thread_id)
// if _, ok := thread_to_attachment_mapping[int(worker_thread_id)]; !ok {
// api.LogDebugf("need to add new thread to the map")
// map_size := len(attachment_to_thread_mapping)
// if map_size < len(attachments_map) {
// attachment_to_thread_mapping[map_size] = worker_thread_id
// thread_to_attachment_mapping[worker_thread_id] = map_size
// api.LogDebugf("len(attachment_to_thread_mapping): %d", len(attachment_to_thread_mapping))
// api.LogDebugf("thread_to_attachment_mapping: %v", thread_to_attachment_mapping)
// api.LogDebugf("attachment_to_thread_mapping: %v", attachment_to_thread_mapping)
// } else {
// panic("unexpected thread id")
// }
// }
// worker_id := thread_to_attachment_mapping[int(worker_thread_id)]
// api.LogDebugf("worker_id: %d", worker_id)
// filter_id.Add(1)
// session_id := filter_id.Load()
// attachment_ptr := attachments_map[worker_id]
// session_data := C.InitSessionData((*C.NanoAttachment)(attachment_ptr), C.SessionID(session_id))
// return &filter{
// callbacks: callbacks,
// config: conf,
// session_id: session_id,
// cp_attachment: attachment_ptr,
// session_data: session_data,
// request_structs: attachment_to_filter_request_structs[worker_id],
// }
// }
// }
func ConfigFactory(c interface{}, callbacks api.FilterCallbackHandler) api.StreamFilter {
func ConfigFactory(c interface{}) api.StreamFilterFactory {
conf, ok := c.(*config)
if !ok {
panic("unexpected config type")
}
worker_thread_id := int(C.get_thread_id())
api.LogDebugf("worker_thread_id: %d", worker_thread_id)
if _, ok := thread_to_attachment_mapping[int(worker_thread_id)]; !ok {
api.LogDebugf("need to add new thread to the map")
map_size := len(attachment_to_thread_mapping)
if map_size < len(attachments_map) {
attachment_to_thread_mapping[map_size] = worker_thread_id
thread_to_attachment_mapping[worker_thread_id] = map_size
api.LogDebugf("len(attachment_to_thread_mapping): %d", len(attachment_to_thread_mapping))
api.LogDebugf("thread_to_attachment_mapping: %v", thread_to_attachment_mapping)
api.LogDebugf("attachment_to_thread_mapping: %v", attachment_to_thread_mapping)
} else {
panic("unexpected thread id")
return func(callbacks api.FilterCallbackHandler) api.StreamFilter {
worker_thread_id := int(C.get_thread_id())
api.LogDebugf("worker_thread_id: %d", worker_thread_id)
if _, ok := thread_to_attachment_mapping[int(worker_thread_id)]; !ok {
api.LogDebugf("need to add new thread to the map")
map_size := len(attachment_to_thread_mapping)
if map_size < len(attachments_map) {
attachment_to_thread_mapping[map_size] = worker_thread_id
thread_to_attachment_mapping[worker_thread_id] = map_size
api.LogDebugf("len(attachment_to_thread_mapping): %d", len(attachment_to_thread_mapping))
api.LogDebugf("thread_to_attachment_mapping: %v", thread_to_attachment_mapping)
api.LogDebugf("attachment_to_thread_mapping: %v", attachment_to_thread_mapping)
} else {
panic("unexpected thread id")
}
}
}
worker_id := thread_to_attachment_mapping[int(worker_thread_id)]
api.LogDebugf("worker_id: %d", worker_id)
worker_id := thread_to_attachment_mapping[int(worker_thread_id)]
api.LogDebugf("worker_id: %d", worker_id)
filter_id.Add(1)
session_id := filter_id.Load()
attachment_ptr := attachments_map[worker_id]
session_data := C.InitSessionData((*C.NanoAttachment)(attachment_ptr), C.SessionID(session_id))
filter_id.Add(1)
session_id := filter_id.Load()
attachment_ptr := attachments_map[worker_id]
session_data := C.InitSessionData((*C.NanoAttachment)(attachment_ptr), C.SessionID(session_id))
return &filter{
callbacks: callbacks,
config: conf,
session_id: session_id,
cp_attachment: attachment_ptr,
session_data: session_data,
request_structs: attachment_to_filter_request_structs[worker_id],
return &filter{
callbacks: callbacks,
config: conf,
session_id: session_id,
cp_attachment: attachment_ptr,
session_data: session_data,
request_structs: attachment_to_filter_request_structs[worker_id],
}
}
}
func main() {}

43
attachments/envoy/filter.go
View File

@@ -107,6 +107,7 @@ type filter struct {
session_data *C.HttpSessionData
cp_attachment *nano_attachment
request_structs *filterRequestStructs
body_buffer_chunk int
}
type filterRequestStructs struct {
@@ -230,10 +231,6 @@ func (f *filter) sendBody(buffer api.BufferInstance, is_req bool) C.AttachmentVe
data := buffer.Bytes()
data_len := len(data)
buffer_size := 8 * 1024
// body_chunk := newNanoStr(data)
// body_chunk.data = (*C.uchar)(unsafe.Pointer(&data[0]))
num_of_buffers := ((data_len - 1) / buffer_size) + 1
// TO DO: FIX THIS ASAP
@@ -299,20 +296,7 @@ func (f *filter) handleStartTransaction(header api.RequestHeaderMap) {
}
func (f *filter) sendLocalReplyInternal(ret_code int, custom_response string, headers map[string][]string) api.StatusType {
//f.callbacks.DecoderFilterCallbacks().SendLocalReply(ret_code, custom_response, headers, 0, "") // new api
// var headers_map map[string]string = nil
// if headers != nil {
// headers_map = make(map[string]string)
// for key, val := range headers {
// header_val := ""
// if len(val) > 0 {
// header_val = val[0]
// }
// headers_map[key] = header_val
// }
// }
f.callbacks.DecoderFilterCallbacks().SendLocalReply(ret_code, custom_response, headers, 0, "")
f.callbacks.SendLocalReply(ret_code, custom_response, headers, 0, "")
return api.LocalReply
}
@@ -428,6 +412,21 @@ func (f *filter) EncodeHeaders(header api.ResponseHeaderMap, endStream bool) api
return ret
}
func injectBodyChunk(
curr_modification *C.struct_NanoHttpModificationList,
body_buffer_chunk int,
buffer *api.BufferInstance) {
for curr_modification != nil {
if (int(curr_modification.modification.orig_buff_index) == body_buffer_chunk) {
mod := curr_modification.modification // type: HttpInjectData
modifications := C.GoString(curr_modification.modification_buffer)
new_buffer:= insertAtPosition((*buffer).String(), modifications, int(mod.injection_pos))
(*buffer).SetString(new_buffer)
}
curr_modification = curr_modification.next
}
}
// EncodeData might be called multiple times during handling the response body.
// The endStream is true when handling the last piece of the body.
func (f *filter) EncodeData(buffer api.BufferInstance, endStream bool) api.StatusType {
@@ -448,6 +447,8 @@ func (f *filter) EncodeData(buffer api.BufferInstance, endStream bool) api.Statu
}
res := f.sendBody(buffer, false)
injectBodyChunk(res.modifications, f.body_buffer_chunk, &buffer)
f.body_buffer_chunk++
if C.AttachmentVerdict(res.verdict) != C.ATTACHMENT_VERDICT_INSPECT {
api.LogInfof("got final verict: %v", res.verdict)
return f.finalizeRequest(&res)
@@ -471,15 +472,15 @@ func (f *filter) EncodeTrailers(trailers api.ResponseTrailerMap) api.StatusType
}
// OnLog is called when the HTTP stream is ended on HTTP Connection Manager filter.
func (f *filter) OnLog(api.RequestHeaderMap, api.RequestTrailerMap, api.ResponseHeaderMap, api.ResponseTrailerMap) {}
func (f *filter) OnLog() {}
// OnLogDownstreamStart is called when HTTP Connection Manager filter receives a new HTTP request
// (required the corresponding access log type is enabled)
func (f *filter) OnLogDownstreamStart(api.RequestHeaderMap) {}
func (f *filter) OnLogDownstreamStart() {}
// OnLogDownstreamPeriodic is called on any HTTP Connection Manager periodic log record
// (required the corresponding access log type is enabled)
func (f *filter) OnLogDownstreamPeriodic(api.RequestHeaderMap, api.RequestTrailerMap, api.ResponseHeaderMap, api.ResponseTrailerMap) {}
func (f *filter) OnLogDownstreamPeriodic() {}
func (f *filter) OnDestroy(reason api.DestroyReason) {
freeHttpMetaDataFields(f.request_structs.http_meta_data)

8
attachments/envoy/go.mod
View File

@@ -1,15 +1,13 @@
module gitlab.ngen.checkpoint.com/Ngen/agent-core/attachments/envoy
// the version should >= 1.18
go 1.22
toolchain go1.22.5
go 1.20
// NOTICE: these lines could be generated automatically by "go mod tidy"
require (
github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa
github.com/envoyproxy/envoy v1.32.1
google.golang.org/protobuf v1.35.1
github.com/envoyproxy/envoy v1.29.8-0.20240702140355-f3e7e90ed021
google.golang.org/protobuf v1.34.2
)
require github.com/go-chi/chi/v5 v5.1.0

13
attachments/envoy/go.sum
View File

@@ -1,7 +1,11 @@
github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa h1:jQCWAUqqlij9Pgj2i/PB79y4KOPYVyFYdROxgaCwdTQ=
github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa/go.mod h1:x/1Gn8zydmfq8dk6e9PdstVsDgu9RuyIIJqAaF//0IM=
github.com/envoyproxy/envoy v1.32.1 h1:+HeajIC+S9PH3mjY/bVqJabjprqxA7h6pSQ+Ie1Ziww=
github.com/envoyproxy/envoy v1.32.1/go.mod h1:KGS+IUehDX1mSIdqodPTWskKOo7bZMLLy3GHxvOKcJk=
github.com/envoyproxy/envoy v1.28.1-0.20231218050644-ca5d45d1887a h1:PQveCjvjXZS400K7z+uuouN/Q/dLhLpB5a/6GIC4v34=
github.com/envoyproxy/envoy v1.28.1-0.20231218050644-ca5d45d1887a/go.mod h1:STO/nOGQMw2DNOFTaokM1VY72GJBs0mbXq0I1lJr0XQ=
github.com/envoyproxy/envoy v1.29.8-0.20240702140355-f3e7e90ed021 h1:mTisjPVHGpxlWi7Yj7PnpxD0GeoauHcWvEgch069i+M=
github.com/envoyproxy/envoy v1.29.8-0.20240702140355-f3e7e90ed021/go.mod h1:ujBFxE543X8OePZG+FbeR9LnpBxTLu64IAU7A20EB9A=
github.com/envoyproxy/envoy v1.31.0 h1:NsTo+medzu0bMffXAjl+zKaViLOShKuIZWQnKKYq0/4=
github.com/envoyproxy/envoy v1.31.0/go.mod h1:ujBFxE543X8OePZG+FbeR9LnpBxTLu64IAU7A20EB9A=
github.com/envoyproxy/protoc-gen-validate v1.0.2 h1:QkIBuU5k+x7/QXPvPPnWXWlCdaBFApVqftFV6k087DA=
github.com/envoyproxy/protoc-gen-validate v1.0.2/go.mod h1:GpiZQP3dDbg4JouG/NNS7QWXpgx6x8QiMKdmN72jogE=
github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw=
@@ -11,7 +15,6 @@ github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg
github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY=
github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/genproto/googleapis/api v0.0.0-20240102182953-50ed04b92917 h1:rcS6EyEaoCO52hQDupoSfrxI3R6C2Tq741is7X8OvnM=
google.golang.org/genproto/googleapis/api v0.0.0-20240102182953-50ed04b92917/go.mod h1:CmlNWB9lSezaYELKS5Ym1r44VrrbPUa7JTvw+6MbpJ0=
@@ -19,5 +22,5 @@ google.golang.org/genproto/googleapis/rpc v0.0.0-20240102182953-50ed04b92917 h1:
google.golang.org/genproto/googleapis/rpc v0.0.0-20240102182953-50ed04b92917/go.mod h1:xtjpI3tXFPP051KaWnhvxkiubL/6dJ18vLVf7q2pTOU=
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
google.golang.org/protobuf v1.34.2 h1:6xV6lTsCfpGD21XK49h7MhtcApnLqkfYgPcdHftf6hg=
google.golang.org/protobuf v1.34.2/go.mod h1:qYOHts0dSfpeUzUFpOMr/WGzszTmLH+DiWniOlNbLDw=

36
attachments/envoy/utils.go
View File

@@ -14,6 +14,7 @@ import (
"unsafe"
"os"
"runtime"
"strconv"
)
func getEnv(key, defaultValue string) string {
value, exists := os.LookupEnv(key)
@@ -23,49 +24,48 @@ func getEnv(key, defaultValue string) string {
return value
}
var INSERT_POS_ERR_MSG = "Got invalid insertion position, will not insert."
func copyToSlice(dest []byte, src unsafe.Pointer, size C.size_t, location int) int {
C.memcpy(unsafe.Pointer(&dest[location]), src, size)
return location + int(size)
}
func newNanoStr(data []byte) *C.nano_str_t {
// Allocate memory for the nano_str_t struct
nanoStr := (*C.nano_str_t)(C.malloc(C.size_t(unsafe.Sizeof(C.nano_str_t{}))))
if nanoStr == nil {
panic("failed to allocate memory for nano_str_t struct")
}
// Set the length of the data
nanoStr.len = C.size_t(len(data))
// Allocate memory for the data field and copy the Go byte slice into it
// nanoStr.data = (*C.uchar)(C.malloc(C.size_t(len(data))))
// if nanoStr.data == nil {
// C.free(unsafe.Pointer(nanoStr))
// panic("failed to allocate memory for data field")
// }
// copy((*[1 << 30]byte)(unsafe.Pointer(nanoStr.data))[:len(data):len(data)], data)
return nanoStr
}
func insertAtPosition(buff string, injection string, pos int) string {
if pos < 0 || pos > len(buff) {
api.LogDebugf(
INSERT_POS_ERR_MSG +
" Position: " +
strconv.Itoa(pos) +
", buffer's lenght: " +
strconv.Itoa(len(buff)))
return buff
}
return_buff := buff[:pos] + injection + buff[pos:]
return return_buff
}
func createNanoStr(str string) C.nano_str_t {
c_str := C.CString(str)
// nanoStr := (*C.nano_str_t)(C.malloc(C.size_t(unsafe.Sizeof(C.nano_str_t{}))))
nanoStr := C.nano_str_t{
len: C.size_t(len(str)),
data: (*C.uchar)(unsafe.Pointer(c_str)),
}
// nanoStr.len = C.size_t(len(str))
// nanoStr.data = (*C.uchar)(unsafe.Pointer(c_str))
return nanoStr
}
func createNanoStrWithoutCopy(str string) C.nano_str_t {
//c_str := C.CString(str)
nanoStr := C.nano_str_t{
len: C.size_t(len(str)),
data: (*C.uchar)(unsafe.Pointer((*(*reflect.StringHeader)(unsafe.Pointer(&str))).Data)),
@@ -105,4 +105,4 @@ func RecoverPanic(ret *api.StatusType) {
*ret = api.Continue
}
}
}