使用github.com/gopcua/opcua连接KEPServer遇身份令牌无效错误

问题描述 #

我尝试用github.com/gopcua/opcua库连接KEPServer，始终收到错误：The user identity token is not valid. StatusBadIdentityTokenInvalid (0x80200000)。我的Go代码已经明确设置SecurityMode为MessageSecurityModeNone、SecurityPolicy为http://opcfoundation.org/UA/SecurityPolicy#None，和目标服务器的安全配置完全匹配，但相同的用户名密码在Python的opcua库中能正常连接服务器。

我的Go实现代码（原版本） #

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"

    "github.com/gopcua/opcua"
    "github.com/gopcua/opcua/id"
    "github.com/gopcua/opcua/ua"
    "github.com/xuri/excelize/v2"
)

var (
    username = *****   // 定义为字符串变量
    password = ***** // 定义为字符串变量
)

const (
    serverAddress  = ********                  // 替换为正确的服务器地址
    applicationURI = "urn:WIN-TQJFKOH9BP4:KEPServerEX.V6:UA%20Server" // 应用URI
)

type NodeData struct {
    NodeID string
    Value  string
}

func main() {
    ctx := context.Background()

    // 发现可用的OPC UA服务器
    fmt.Printf("正在%s处发现服务器...\n", serverAddress)
    servers, err := opcua.FindServers(ctx, serverAddress)
    if err != nil {
        log.Fatalf("发现服务器失败: %v", err)
    }

    if len(servers) == 0 {
        log.Fatal("未找到任何服务器")
    }

    // 打印发现的服务器
    fmt.Println("已发现服务器:")
    for i, server := range servers {
        fmt.Printf("服务器 %d: %s\n", i+1, server.ApplicationName.Text) // 使用ApplicationName.Text
        for _, url := range server.DiscoveryURLs {
            fmt.Printf("  发现URL: %s\n", url)
        }
        fmt.Println("----------------------------------------")
    }

    // 获取第一个服务器的端点
    discoveryURL := serverAddress
    fmt.Printf("正在从%s获取端点...\n", discoveryURL)
    endpoints, err := opcua.GetEndpoints(ctx, discoveryURL)
    if err != nil {
        log.Fatalf("获取端点失败: %v", err)
    }

    if len(endpoints) == 0 {
        log.Fatal("未找到任何端点")
    }

    // 打印发现的端点
    fmt.Println("已发现端点:")
    for i, endpoint := range endpoints {
        fmt.Printf("端点 %d: %s\n", i+1, endpoint.EndpointURL)
        fmt.Printf("  安全模式: %s\n", endpoint.SecurityMode)
        fmt.Printf("  安全策略: %s\n", endpoint.SecurityPolicyURI)
        fmt.Printf("  传输配置文件: %s\n", endpoint.TransportProfileURI)
        fmt.Println("----------------------------------------")
    }

    // 连接到第6个端点（索引5）
    if len(endpoints) < 1 {
        log.Fatalf("可用端点数量不足。预期至少6个，实际得到%d个", len(endpoints))
    }
    endpoint := serverAddress
    fmt.Printf("\n正在连接到端点6: %s\n", endpoint)
    client := connectWithRetry(ctx, endpoint, username, password, applicationURI) // 传入用户名、密码和应用URI
    defer client.Close(ctx)

    var wg sync.WaitGroup
    var mu sync.Mutex
    var nodes []NodeData
    queue := make(chan *ua.NodeID, 100)

    queue <- ua.NewNumericNodeID(0, id.ObjectsFolder) // 从Objects文件夹开始浏览

    for i := 0; i < 5; i++ { // 使用5个工作协程
        wg.Add(1)
        go func() {
            defer wg.Done()
            browseNodes(ctx, client, queue, &nodes, &mu)
        }()
    }

    wg.Wait()
    close(queue)
    saveToExcel(nodes)
    fmt.Println("数据已保存到OPC_Data.xlsx")
}

// 更新后的函数，接受用户名、密码和应用URI
func connectWithRetry(ctx context.Context, endpoint, username, password, applicationURI string) *opcua.Client {
    for {
        client, err := opcua.NewClient(endpoint,
            opcua.SecurityPolicy(ua.SecurityPolicyURINone),             // 无安全策略
            opcua.SecurityMode(ua.MessageSecurityModeNone),             // 无安全模式
            opcua.AuthUsername(username, password), // 用户名密码认证
            // Set Application URI
            opcua.AuthPolicyID("username"), // 显式设置UserTokenType为UserName
        )

        if err != nil {
            log.Println("客户端创建失败，5秒后重试...", err)
            time.Sleep(5 * time.Second)
            continue
        }
        if err := client.Connect(ctx); err != nil {
            log.Println("连接失败，5秒后重试...", err)
            time.Sleep(5 * time.Second)
        } else {
            fmt.Println("✅ 已连接到OPC UA服务器")
            return client
        }
    }
}

func browseNodes(ctx context.Context, client *opcua.Client, queue chan *ua.NodeID, nodes *[]NodeData, mu *sync.Mutex) {
    for nodeID := range queue {
        resp, err := client.Browse(ctx, &ua.BrowseRequest{
            RequestHeader: new(ua.RequestHeader),
            NodesToBrowse: []*ua.BrowseDescription{{
                NodeID:          nodeID,
                BrowseDirection: ua.BrowseDirectionForward,
                ReferenceTypeID: ua.NewNumericNodeID(0, id.References),
                IncludeSubtypes: true,
                ResultMask:      uint32(ua.BrowseResultMaskAll),
            }},
        })
        if err != nil || resp == nil || len(resp.Results) == 0 || resp.Results[0] == nil || resp.Results[0].References == nil {
            log.Println("浏览失败或响应为空，节点:", nodeID, "错误:", err)
            continue
        }

        for _, ref := range resp.Results[0].References {
            if ref.NodeID == nil || ref.NodeID.NodeID == nil {
                continue
            }

            if ref.NodeClass == ua.NodeClassVariable {
                val, err := readNodeValue(ctx, client, ref.NodeID.NodeID)
                if err == nil {
                    mu.Lock()
                    *nodes = append(*nodes, NodeData{NodeID: ref.NodeID.NodeID.String(), Value: val})
                    mu.Unlock()
                }
            } else {
                select {
                case queue <- ref.NodeID.NodeID:
                default:
                    log.Println("队列已满，丢弃节点:", ref.NodeID.NodeID)
                }
            }
        }
    }
}

func readNodeValue(ctx context.Context, client *opcua.Client, nodeID *ua.NodeID) (string, error) {
    resp, err := client.Read(ctx, &ua.ReadRequest{
        NodesToRead: []*ua.ReadValueID{{NodeID: nodeID, AttributeID: ua.AttributeIDValue}},
    })
    if err != nil || len(resp.Results) == 0 || resp.Results[0].Status != ua.StatusOK {
        return "", fmt.Errorf("读取错误")
    }
    return fmt.Sprintf("%v", resp.Results[0].Value.Value()), nil
}

报错信息 #

正在连接到端点6: ***

2025/03/08 21:48:52 连接失败，5秒后重试... The user identity token is not valid. StatusBadIdentityTokenInvalid (0x80200000)

目标服务器安全配置 #

• Security Mode: MessageSecurityModeNone
• Security Policy: http://opcfoundation.org/UA/SecurityPolicy#None
• Transport Profile: http://opcfoundation.org/UA-Profile/Transport/uatcp-uasc-uabinary

可正常运行的Python代码 #

import sys
from opcua import Client

# OPC UA服务器详情
# 替换为你要读取的实际节点ID

# 连接OPC UA服务器的函数
def connect_opcua(url, username, password):
    client = Client(url)
    try:
        client.set_user(username)
        client.set_password(password)
        client.connect()
        print("✅ 已连接到OPC UA服务器")
        return client
    except Exception as e:
        print(f"❌ 连接失败: {e}")
        sys.exit(1)

# 带重试机制读取节点值的函数
def read_node_value(client, node_id):
    max_retries = 3
    for retry in range(max_retries):  # 合理的重试循环
        try:
            node = client.get_node(node_id)
            value = node.get_value()
            if value is not None:
                print(f"📢 节点ID: {node_id}, 值: {value}")
                return node_id, value
            else:
                print(f"⚠️ 第 {retry + 1} 次尝试: 节点值为空，正在重试...")
        except Exception as e:
            print(f"🚨 第 {retry + 1} 次尝试: 读取节点 {node_id} 出错: {e}")

    print(f"❌ 节点 {node_id} 已达到最大重试次数，返回None")
    return node_id, None  # 所有重试失败后返回None

# 主执行逻辑
if __name__ == "__main__":
    client = connect_opcua(OPC_SERVER_URL, USERNAME, PASSWORD)
    try:
        node_id, value = read_node_value(client, NODE_ID)
    finally:
        client.disconnect()
        print("✅ 已断开与OPC UA服务器的连接")

Python代码运行输出 #

Requested session timeout to be 3600000ms, got 60000ms instead
d:\Mohit\webapp\Streamlit\venv\Lib\site-packages\opcua\crypto\uacrypto.py:27: CryptographyDeprecationWarning: Parsed a negative serial number, which is disallowed by RFC 5280. Loading this certificate will cause an exception in the next release of cryptography.
  return x509.load_der_x509_certificate(data, default_backend())
✅ 已连接到OPC UA服务器
🚨 第1次尝试: 读取节点 ns=2;s=INVERTER.ICR1.LT2.INV1.PV10_A 出错: 
📢 节点ID: ns=2;s=INVERTER.ICR1.LT2.INV1.PV10_A, 值: 1.24
✅ 已断开与OPC UA服务器的连接

修正后的Go代码（关键修复点） #

// ... 其他代码不变 ...

func main() {
    // ... 发现服务器和端点的代码不变 ...

    // 连接到第6个端点（索引5）
    if len(endpoints) < 6 {
        log.Fatalf("可用端点数量不足。预期至少6个，实际得到%d个", len(endpoints))
    }
    // 修复：使用发现到的第6个端点的实际URL，而非发现URL
    endpoint := endpoints[5].EndpointURL
    fmt.Printf("\n正在连接到端点6: %s\n", endpoint)
    client := connectWithRetry(ctx, endpoint, username, password, applicationURI)
    defer client.Close(ctx)

    // ... 后续浏览和保存代码不变 ...
}

func connectWithRetry(ctx context.Context, endpoint, username, password, applicationURI string) *opcua.Client {
    for {
        client, err := opcua.NewClient(endpoint,
            opcua.SecurityPolicy(ua.SecurityPolicyURINone),
            opcua.SecurityMode(ua.MessageSecurityModeNone),
            opcua.AuthUsername(username, password),
            // 修复：添加应用URI设置，部分服务器会验证该字段
            opcua.ApplicationURI(applicationURI),
            // 可选：尝试省略AuthPolicyID，让库自动协商服务器支持的策略
            // opcua.AuthPolicyID("username"),
        )

        // ... 重试逻辑不变 ...
    }
}

// ... 其他函数不变 ...

关键修复说明 #

1. 使用正确的端点URL：原代码错误地将发现URL（serverAddress）作为连接地址，实际应使用发现到的第6个端点的EndpointURL，KEPServer的发现URL和业务端点URL可能不同，导致身份验证失败。
2. 添加应用URI设置：显式传入opcua.ApplicationURI(applicationURI)，确保客户端提供服务器期望的应用标识，部分严格的OPC UA服务器会验证该字段。
3. 可选：调整AuthPolicyID：如果服务器的用户名认证策略ID不是"username"，可以尝试省略该选项，让库自动协商服务器支持的策略。

内容的提问来源于stack exchange，提问作者MOHIT KHARE