GO實現(xiàn)基于命令行的簡單IPS程序代碼

更新時間：2024年12月28日 11:21:03 作者：星星貓

本文介紹了入侵防御系統(tǒng)IPS的工作原理和實現(xiàn),IPS通過網(wǎng)絡(luò)流量監(jiān)控和實時響應(yīng),防止網(wǎng)絡(luò)攻擊,通過使用Go語言實現(xiàn)一個簡單的IPS示例程序,展示了如何獲取本地IP地址和探測網(wǎng)絡(luò)中其他設(shè)備的IP地址,包括如何定義和加載規(guī)則文件,以及如何檢測IP對相應(yīng)端口的訪問是否達(dá)到規(guī)定閾值

入侵防御系統(tǒng)

IPS（Intrusion Prevention System）即入侵防御系統(tǒng)，主要用于實時檢查和阻止網(wǎng)絡(luò)入侵。與入侵檢測系統(tǒng)（IDS）不同，IPS通常部署在網(wǎng)絡(luò)的關(guān)鍵路徑上，實時監(jiān)控網(wǎng)絡(luò)流量，并在發(fā)現(xiàn)攻擊時立即采取響應(yīng)措施，如丟棄惡意數(shù)據(jù)包、封鎖攻擊源IP地址等。

IPS的工作原理

IPS通過直接嵌入到網(wǎng)絡(luò)流量中實現(xiàn)主動防御。它通過一個網(wǎng)絡(luò)端口接收來自外部系統(tǒng)的流量，經(jīng)過檢查確認(rèn)其中不包含異?；顒踊蚩梢蓛?nèi)容后，再通過另一個端口傳送到內(nèi)部系統(tǒng)中。有問題的數(shù)據(jù)包以及所有來自同一數(shù)據(jù)流的后續(xù)數(shù)據(jù)包都會在IPS設(shè)備中被清除掉。

GO實現(xiàn)IPS

下面我們用 Go 語言編寫，一個可以在命令行中獲取并打印本地 IP 地址（基于常見網(wǎng)絡(luò)接口）的示例程序，這里主要是獲取 IPv4 地址進(jìn)行展示，示例代碼使用了標(biāo)準(zhǔn)庫中的 net 包：

獲取本地網(wǎng)絡(luò)接口上綁定的 IP

package main

import (
    "fmt"
    "net"
)

func main() {
    addrs, err := net.InterfaceAddrs()
    if err!= nil {
        fmt.Printf("獲取網(wǎng)絡(luò)接口地址失敗: %v\n", err)
        return
    }

    for _, addr := range addrs {
        ipNet, ok := addr.(*net.IPNet)
        if ok &&!ipNet.IP.IsLoopback() && ipNet.IP.To4()!= nil {
            fmt.Println("IP地址:", ipNet.IP.String())
        }
    }
}

運行

確保已經(jīng)安裝了 Go 環(huán)境（Go 1.10 及以上版本）。
將上述代碼保存到一個以 .go 為后綴的文件中，比如 ips.go。
打開命令行，切換到該文件所在的目錄。
運行命令 go run ips.go，程序就會執(zhí)行并輸出找到的本地非回環(huán)的 IPv4 地址信息。

也可編譯后運行

go build ips.go

然后運行生成的可執(zhí)行文件（在 Windows 上是 .exe 后綴的文件，在 Linux、macOS 等系統(tǒng)上就是普通的二進(jìn)制文件）也能看到相應(yīng)的 IP 地址輸出。

請注意：這里只是簡單獲取本地網(wǎng)絡(luò)接口上綁定的 IP 示例

對于IPS而言，他要檢測的是網(wǎng)絡(luò)中的流量（比如探測網(wǎng)絡(luò)中其他設(shè)備 IP 等）這需要進(jìn)一步擴(kuò)展代碼，例如利用 net.Dial 等去做網(wǎng)絡(luò)連接測試、IP 掃描等操作。

探測網(wǎng)絡(luò)中其他設(shè)備 IP

這里只是通過簡單的 ICMP 即 ping 操作概念類似的簡單探測

package main

import (
    "fmt"
    "net"
    "os"
    "sync"
    "time"
)

func PingIP(ip string, wg *sync.WaitGroup, result chan<- string) {
    defer wg.Done()
    conn, err := net.DialTimeout("ip4:icmp", ip, time.Second*2)
    if err == nil {
        conn.Close()
        result <- ip
    }
}

func main() {
    if len(os.Args)!= 2 {
        fmt.Println("請輸入要掃描的網(wǎng)段，例如 192.168.1.0/24")
        return
    }
    _, ipnet, err := net.ParseCIDR(os.Args[1])
    if err!= nil {
        fmt.Printf("解析網(wǎng)段失敗: %v\n", err)
        return
    }

    var wg sync.WaitGroup
    result := make(chan string)

    for ip := ipnet.IP.Mask(ipnet.Mask); ipnet.Contains(ip); inc(ip) {
        wg.Add(1)
        go PingIP(ip.String(), &wg, result)
    }

    go func() {
        wg.Wait()
        close(result)
    }()

    for aliveIP := range result {
        fmt.Println("存活I(lǐng)P:", aliveIP)
    }
}

func inc(ip net.IP) {
    for j := len(ip) - 1; j >= 0; j-- {
        ip[j]++
        if ip[j] > 0 {
            break
        }
    }
}

在命令行中使用這個擴(kuò)展版本時，可以像這樣運行（假設(shè)編譯后的可執(zhí)行文件名為 ips）：

./ips 192.168.1.0/24

上述命令會嘗試去探測指定網(wǎng)段 192.168.1.0/24 內(nèi)的 IP 地址哪些是存活的（可以響應(yīng) ICMP 請求的，類似 ping 通的概念），并打印出存活的 IP 地址。

增加規(guī)則文件（rules.json）

首先我們設(shè)定一個規(guī)則文件，用json的方式來保存規(guī)則

[
    {
        "id": "1001",
        "description": "SSH暴力破解檢測",
        "protocol": "tcp",
        "dst_port": 22,
        "action": "block",
        "threshold": 5,
        "time_window": 60
    },
    {
        "id": "1002",
        "description": "RDP暴力破解檢測",
        "protocol": "tcp",
        "dst_port": 3389,
        "action": "block",
        "threshold": 5,
        "time_window": 60
    },
    {
        "id": "1003",
        "description": "MySQL暴力破解檢測",
        "protocol": "tcp",
        "dst_port": 3306,
        "action": "block",
        "threshold": 5,
        "time_window": 60
    }
]

然后我們寫一個go文件，來檢測 IP 對相應(yīng)端口的訪問是否達(dá)到規(guī)則中定義的暴力破解閾值情況

package main

import (
    "encoding/json"
    "fmt"
    "io/ioutil"
    "net"
    "os"
    "path/filepath"
    "sync"
    "time"
)

// Rule結(jié)構(gòu)體表示一條安全規(guī)則
type Rule struct {
    ID          string  `json:"id"`
    Description string  `json:"description"`
    Protocol    string  `json:"protocol"`
    DstPort     int     `json:"dst_port"`
    Action      string  `json:"action"`
    Threshold   int     `json:"threshold"`
    TimeWindow  int     `json:"time_window"`
}

// LoadRules從文件加載安全規(guī)則
func LoadRules(filePath string) ([]*Rule, error) {
    content, err := ioutil.ReadFile(filePath)
    if err!= nil {
        return nil, fmt.Errorf("讀取規(guī)則文件失敗: %v", err)
    }
    var rules []*Rule
    err = json.Unmarshal(content, &rules)
    if err!= nil {
        return nil, fmt.Errorf("解析規(guī)則文件失敗: %v", err)
    }
    return rules, nil
}

// ConnectionRecord記錄IP的連接記錄（簡化示例，僅記錄次數(shù)）
type ConnectionRecord struct {
    mu         sync.Mutex
    connections map[string]int
}

func NewConnectionRecord() *ConnectionRecord {
    return &ConnectionRecord{
        connections: make(map[string]int),
    }
}

func (cr *ConnectionRecord) Record(ip net.IP, port int) {
    cr.mu.Lock()
    key := fmt.Sprintf("%s:%d", ip.String(), port)
    cr.connections[key]++
    cr.mu.Unlock()
}

func (cr *ConnectionRecord) CheckViolation(rules []*Rule) []*Rule {
    var violatedRules []*Rule
    cr.mu.Lock()
    now := time.Now()
    for _, rule := range rules {
        for key, count := range cr.connections {
            parts := strings.Split(key, ":")
            ip := net.ParseIP(parts[0])
            port := atoi(parts[1])
            if port == rule.DstPort && count >= rule.Threshold {
                // 這里簡單判斷次數(shù)達(dá)到閾值就算違規(guī)，實際可能需考慮時間窗口等更復(fù)雜邏輯細(xì)化
                violatedRules = append(violatedRules, rule)
            }
        }
    }
    cr.mu.Unlock()
    return violatedRules
}

func atoi(s string) int {
    n, _ := fmt.Sscanf(s, "%d", &n)
    return n
}

func main() {
    // 獲取當(dāng)前目錄下的rules.json作為配置文件路徑（可根據(jù)實際調(diào)整）
    currentDir, err := os.Getwd()
    if err!= nil {
        fmt.Printf("獲取當(dāng)前目錄失敗: %v\n", err)
        return
    }
    configFilePath := filepath.Join(currentDir, "rules.json")

    rules, err := LoadRules(configFilePath)
    if err!= nil {
        fmt.Printf("加載規(guī)則失敗: %v\n", err)
        return
    }

    targetIPNet := &net.IPNet{}
    _, targetIPNet, err = net.ParseCIDR("192.168.1.0/24")
    if err!= nil {
        fmt.Printf("解析目標(biāo)IP網(wǎng)段失敗: %v\n", err)
        return
    }

    record := NewConnectionRecord()

    // 模擬IP對內(nèi)網(wǎng)端口的連接訪問（這里簡單循環(huán)增加連接次數(shù)示例）
    for i := 0; i < 10; i++ {
        for ip := targetIPNet.IP.Mask(targetIPNet.Mask); targetIPNet.Contains(ip); inc(ip) {
            // 簡單模擬多次訪問各規(guī)則關(guān)注的端口，實際場景會根據(jù)真實網(wǎng)絡(luò)連接情況記錄
            for _, rule := range rules {
                record.Record(ip, rule.DstPort)
            }
        }
    }

    violatedRules := record.CheckViolation(rules)
    for _, violatedRule := range violatedRules {
        fmt.Printf("IP段內(nèi)存在疑似違反規(guī)則 %s 的情況，規(guī)則描述：%s\n", violatedRule.ID, violatedRule.Description)
    }
}

func inc(ip net.IP) {
    for j := len(ip) - 1; j >= 0; j-- {
        ip[j]++
        if ip[j] > 0 {
            break
        }
    }
}

定義規(guī)則結(jié)構(gòu)體和相關(guān)操作函數(shù)：

首先定義了 Rule 結(jié)構(gòu)體，其字段與 JSON 格式的規(guī)則文件中的各個屬性相對應(yīng)，用于表示一條安全規(guī)則。
LoadRules 函數(shù)用于讀取指定路徑的 JSON 規(guī)則文件，并將其解析為 Rule 結(jié)構(gòu)體的切片，若讀取或解析過程出現(xiàn)錯誤則返回相應(yīng)的錯誤信息。

定義連接記錄結(jié)構(gòu)體及相關(guān)方法：

ConnectionRecord 結(jié)構(gòu)體用于記錄 IP 地址對不同端口的連接情況，內(nèi)部使用一個互斥鎖（sync.Mutex）來保證并發(fā)安全，通過 Record 方法記錄每個 IP 和端口的連接次數(shù)，使用 CheckViolation 方法來檢查當(dāng)前的連接記錄是否違反了給定的規(guī)則（這里只是簡單根據(jù)連接次數(shù)是否達(dá)到閾值判斷，實際可進(jìn)一步完善考慮時間窗口等更多細(xì)節(jié)）。

主函數(shù)邏輯：

先獲取當(dāng)前目錄下規(guī)則配置文件（rules.json）的路徑（可按需修改為其他路徑），調(diào)用 LoadRules 函數(shù)加載規(guī)則列表。
接著解析要檢測的目標(biāo) IP 網(wǎng)段（示例中為 192.168.1.0/24，可以根據(jù)實際情況更改），創(chuàng)建一個 ConnectionRecord 實例來記錄連接情況。
然后通過循環(huán)模擬 IP 對內(nèi)網(wǎng)各端口的訪問（這里只是簡單循環(huán)增加訪問次數(shù)示意，實際需要結(jié)合真實的網(wǎng)絡(luò)連接情況來記錄），并調(diào)用 ConnectionRecord 的 Record 方法記錄每次的訪問情況。
最后調(diào)用 ConnectionRecord 的 CheckViolation 方法檢查是否有違反規(guī)則的情況，若有則打印出相應(yīng)的違規(guī)規(guī)則信息。