Очередь job получила lease и fencing, телеметрия и прогресс deploy сохраняются, ревизия пишет manifest входов. Обновление префиксов не переписывает неизменённый снимок, большие наборы идут через COPY, а устаревшая ревизия не попадает в deploy. Бандл ноды кэшируется, prune и постановка job больше не держат горячий путь. Co-authored-by: Cursor <[email protected]>
246 lines
6.4 KiB
Go
246 lines
6.4 KiB
Go
package importer
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/csv"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"evobgp/internal/store"
|
|
)
|
|
|
|
// MaxCSVBytes is the complete request limit, including the optional header.
|
|
const MaxCSVBytes = 8 << 20
|
|
|
|
// MaxCSVRows bounds database work for a single import, excluding its header.
|
|
const MaxCSVRows = 10000
|
|
|
|
// MaxCSVFields bounds CSV parser allocations for delimiter-heavy records.
|
|
const MaxCSVFields = 64
|
|
|
|
// ErrCSVTooLarge means the entire import was rejected before creating entries.
|
|
var ErrCSVTooLarge = errors.New("importer: csv exceeds byte, row or field limit")
|
|
|
|
// Result of importing module entries from CSV.
|
|
type Result struct {
|
|
Imported int
|
|
ModuleType string
|
|
}
|
|
|
|
// ImportModuleEntriesCSV parses and creates entries for AS_PREFIXES, DOMAINS, or IP_RANGES modules.
|
|
func ImportModuleEntriesCSV(ctx context.Context, st store.Backend, tenantID, moduleID string, body io.Reader) (*Result, error) {
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
mod, err := st.GetModule(tenantID, moduleID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Read one byte beyond the cap: a valid prefix must never be imported as if
|
|
// it were the complete request. Check all resource limits before mutations.
|
|
raw, err := io.ReadAll(io.LimitReader(body, MaxCSVBytes+1))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("importer: read csv: %w", err)
|
|
}
|
|
if len(raw) > MaxCSVBytes {
|
|
return nil, ErrCSVTooLarge
|
|
}
|
|
if err := checkCSVFields(ctx, raw); err != nil {
|
|
return nil, err
|
|
}
|
|
cr := csv.NewReader(bytes.NewReader(raw))
|
|
cr.TrimLeadingSpace = true
|
|
cr.FieldsPerRecord = -1
|
|
rows := make([][]string, 0)
|
|
headerRows := 0
|
|
for {
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
record, err := cr.Read()
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("importer: invalid csv: %w", err)
|
|
}
|
|
if len(rows) == 0 && isCSVHeader(record) {
|
|
headerRows = 1
|
|
}
|
|
if len(rows)-headerRows >= MaxCSVRows {
|
|
return nil, ErrCSVTooLarge
|
|
}
|
|
rows = append(rows, record)
|
|
}
|
|
if len(rows) == 0 {
|
|
return nil, fmt.Errorf("importer: %w", store.ErrInvalidInput)
|
|
}
|
|
|
|
communities, err := st.ListCommunities(tenantID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
communityIDByID := make(map[string]string, len(communities))
|
|
communityIDByValue := make(map[string]string, len(communities))
|
|
for _, c := range communities {
|
|
communityIDByID[c.ID] = c.ID
|
|
communityIDByValue[strings.TrimSpace(c.Community)] = c.ID
|
|
}
|
|
|
|
resolveCommunity := func(raw string, required bool) (*string, error) {
|
|
v := strings.TrimSpace(raw)
|
|
if v == "" {
|
|
if required {
|
|
return nil, fmt.Errorf("community is required")
|
|
}
|
|
return nil, nil
|
|
}
|
|
if id, ok := communityIDByID[v]; ok {
|
|
return &id, nil
|
|
}
|
|
if id, ok := communityIDByValue[v]; ok {
|
|
return &id, nil
|
|
}
|
|
return nil, fmt.Errorf("unknown community %q", v)
|
|
}
|
|
|
|
start := 0
|
|
if isCSVHeader(rows[0]) {
|
|
start = 1
|
|
}
|
|
|
|
imported := 0
|
|
switch mod.Type {
|
|
case "AS_PREFIXES":
|
|
for i := start; i < len(rows); i++ {
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
rec := rows[i]
|
|
if len(rec) == 0 || (strings.TrimSpace(rec[0]) == "" && (len(rec) < 2 || strings.TrimSpace(rec[1]) == "")) {
|
|
continue
|
|
}
|
|
if len(rec) < 2 {
|
|
return nil, fmt.Errorf("importer: line %d: expected 2 columns", i+1)
|
|
}
|
|
asn, err := strconv.ParseInt(strings.TrimSpace(rec[0]), 10, 64)
|
|
if err != nil || asn <= 0 {
|
|
return nil, fmt.Errorf("importer: line %d: invalid asn", i+1)
|
|
}
|
|
cid, err := resolveCommunity(rec[1], false)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("importer: line %d: %w", i+1, err)
|
|
}
|
|
if _, err := st.CreateASEntry(tenantID, moduleID, &store.ASEntry{ASN: asn, CommunityID: cid}); err != nil {
|
|
return nil, err
|
|
}
|
|
imported++
|
|
}
|
|
case "DOMAINS":
|
|
for i := start; i < len(rows); i++ {
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
rec := rows[i]
|
|
if len(rec) == 0 || (strings.TrimSpace(rec[0]) == "" && (len(rec) < 2 || strings.TrimSpace(rec[1]) == "")) {
|
|
continue
|
|
}
|
|
if len(rec) < 2 {
|
|
return nil, fmt.Errorf("importer: line %d: expected 2 columns", i+1)
|
|
}
|
|
fqdn := strings.TrimSpace(rec[0])
|
|
if fqdn == "" {
|
|
return nil, fmt.Errorf("importer: line %d: domain is required", i+1)
|
|
}
|
|
cid, err := resolveCommunity(rec[1], false)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("importer: line %d: %w", i+1, err)
|
|
}
|
|
if _, err := st.CreateDomainEntry(tenantID, moduleID, &store.DomainEntry{FQDN: fqdn, CommunityID: cid}); err != nil {
|
|
return nil, err
|
|
}
|
|
imported++
|
|
}
|
|
case "IP_RANGES":
|
|
for i := start; i < len(rows); i++ {
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
rec := rows[i]
|
|
if len(rec) == 0 || (strings.TrimSpace(rec[0]) == "" && (len(rec) < 2 || strings.TrimSpace(rec[1]) == "")) {
|
|
continue
|
|
}
|
|
if len(rec) < 2 {
|
|
return nil, fmt.Errorf("importer: line %d: expected 2 columns", i+1)
|
|
}
|
|
prefix := strings.TrimSpace(rec[0])
|
|
if prefix == "" {
|
|
return nil, fmt.Errorf("importer: line %d: ipRange is required", i+1)
|
|
}
|
|
cid, err := resolveCommunity(rec[1], true)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("importer: line %d: %w", i+1, err)
|
|
}
|
|
if _, err := st.CreateIPRangeEntry(tenantID, moduleID, &store.IPRangeEntry{Prefix: prefix, CommunityID: cid}); err != nil {
|
|
return nil, err
|
|
}
|
|
imported++
|
|
}
|
|
default:
|
|
return nil, fmt.Errorf("importer: csv import/export is supported only for AS_PREFIXES, DOMAINS, IP_RANGES")
|
|
}
|
|
|
|
return &Result{Imported: imported, ModuleType: mod.Type}, nil
|
|
}
|
|
|
|
func isCSVHeader(record []string) bool {
|
|
if len(record) < 2 {
|
|
return false
|
|
}
|
|
switch strings.ToLower(strings.TrimSpace(record[0])) {
|
|
case "asn", "domain", "iprange":
|
|
return true
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
|
|
// checkCSVFields runs before encoding/csv allocates a string/index for each field.
|
|
// Quoted commas, escaped quotes and multiline fields don't start new fields.
|
|
// Syntax validation remains the responsibility of encoding/csv below.
|
|
func checkCSVFields(ctx context.Context, raw []byte) error {
|
|
quoted, fields := false, 1
|
|
for i := 0; i < len(raw); i++ {
|
|
if i%4096 == 0 {
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
switch raw[i] {
|
|
case '"':
|
|
if quoted && i+1 < len(raw) && raw[i+1] == '"' {
|
|
i++
|
|
continue
|
|
}
|
|
quoted = !quoted
|
|
case ',':
|
|
if !quoted {
|
|
fields++
|
|
if fields > MaxCSVFields {
|
|
return ErrCSVTooLarge
|
|
}
|
|
}
|
|
case '\n':
|
|
if !quoted {
|
|
fields = 1
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|