feat: implement OPC UA mapper using v1beta1 mapper-framework#138
feat: implement OPC UA mapper using v1beta1 mapper-framework#138Sarvagya035 wants to merge 1 commit intokubeedge:mainfrom
Conversation
Signed-off-by: Sarvagya <sarvagya035@gmail.com>
kubeedge-bot
added
the
kind/feature
|
Welcome @Sarvagya035! It looks like this is your first PR to kubeedge/mappers-go 🎉 |
kubeedge-bot
added
the
size/XXL
|
[APPROVALNOTIFIER] This PR is NOT APPROVED This pull-request has been approved by: The full list of commands accepted by this bot can be found here. DetailsNeeds approval from an approver in each of these files:Approvers can indicate their approval by writing |
Summary of ChangesHello @Sarvagya035, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request delivers a critical update to KubeEdge's device management capabilities by introducing a new OPC UA mapper. This mapper is engineered to be fully compatible with the v1beta1 device management API and the modern mapper-framework, resolving issues faced by users with older v1alpha2-based mappers. The goal is to provide robust and flexible integration for OPC UA devices, enhancing the overall user experience and extending built-in support for industrial automation scenarios within KubeEdge. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new OPC UA mapper built on the v1beta1 mapper-framework. The changes are extensive, adding a complete mapper implementation with drivers, data handlers for various databases and push methods, and associated configuration files. My review focuses on improving security, efficiency, and correctness across the new codebase. I've identified several critical SQL injection vulnerabilities, inefficient resource handling in MQTT and database clients, and potential race conditions or panics. I've provided suggestions to address these issues, such as using parameterized queries, reusing client connections, and ensuring thread safety.
| createTable := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` (id INT AUTO_INCREMENT PRIMARY KEY, ts DATETIME NOT NULL,field TEXT)", tableName) | ||
| _, err := DB.Exec(createTable) | ||
| if err != nil { | ||
| return fmt.Errorf("create tabe into mysql failed with err:%v", err) | ||
| } | ||
|
|
||
| stmt, err := DB.Prepare(fmt.Sprintf("INSERT INTO `%s` (ts,field) VALUES (?,?)", tableName)) | ||
| if err != nil { | ||
| return fmt.Errorf("prepare parament failed with err:%v", err) | ||
| } |
There was a problem hiding this comment.
Constructing SQL queries using fmt.Sprintf with inputs that can be controlled by users (like device metadata) creates a serious SQL injection vulnerability. An attacker could manipulate device names or properties to execute arbitrary SQL commands. Always use parameterized queries for all parts of the SQL statement, including table names if they must be dynamic (though it's better to have a fixed schema). Also, creating the table on every data insertion is inefficient and should be done once during initialization.
// Sanitize table name to prevent SQL injection. Replace invalid characters.
safeTableName := strings.ReplaceAll(tableName, "/", "_")
safeTableName = strings.ReplaceAll(safeTableName, "`", "")
createTable := fmt.Sprintf("CREATE TABLE IF NOT EXISTS `%s` (id INT AUTO_INCREMENT PRIMARY KEY, ts DATETIME NOT NULL, field TEXT)", safeTableName)
_, err := DB.Exec(createTable)
if err != nil {
return fmt.Errorf("create table into mysql failed with err:%v", err)
}
stmt, err := DB.Prepare(fmt.Sprintf("INSERT INTO `%s` (ts,field) VALUES (?,?)", safeTableName))
if err != nil {
return fmt.Errorf("prepare statement failed with err:%v", err)
}| insertSQL := fmt.Sprintf("INSERT INTO %s USING %s TAGS ('%s') VALUES('%v','%s', '%s', '%s', '%s');", | ||
| legalTag, legalTable, legalTag, datatime, tableName, data.PropertyName, data.Value, data.Type) |
There was a problem hiding this comment.
Constructing SQL queries with fmt.Sprintf using potentially user-controlled data (like device and property names) leads to a critical SQL injection vulnerability. An attacker could craft malicious names to execute arbitrary SQL commands. You should always use parameterized queries or strictly sanitize any dynamic parts of the query.
| insertSQL := fmt.Sprintf("INSERT INTO %s USING %s TAGS ('%s') VALUES('%v','%s', '%s', '%s', '%s');", | |
| legalTag, legalTable, legalTag, datatime, tableName, data.PropertyName, data.Value, data.Type) | |
| // It's safer to use parameterized queries if the driver supports them for table/tag names. | |
| // If not, ensure stringent sanitization on legalTag and legalTable. | |
| insertSQL := fmt.Sprintf("INSERT INTO ? USING ? TAGS (?) VALUES(?,?,?,?,?)") | |
| // Then use DB.Exec(insertSQL, legalTag, legalTable, ...) |
| resp, err := http.Post(targetUrl, | ||
| "application/x-www-form-urlencoded", | ||
| strings.NewReader(payload)) | ||
|
|
||
| if err != nil { | ||
| fmt.Println(err) | ||
| } | ||
| defer resp.Body.Close() |
There was a problem hiding this comment.
If http.Post returns an error, resp will be nil. The subsequent defer resp.Body.Close() will cause a panic. The defer statement should be placed after a successful error check. Additionally, the Timeout field in HTTPConfig is not being used. A default http.Client with a configured timeout should be used to prevent the application from hanging on unresponsive requests.
client := http.Client{
Timeout: time.Duration(pm.HTTP.Timeout) * time.Second,
}
resp, err := client.Post(targetUrl,
"application/x-www-form-urlencoded",
strings.NewReader(payload))
if err != nil {
klog.Errorf("HTTP Post error: %v", err)
return
}
defer resp.Body.Close()| RUN CGO_ENABLED=0 GOOS=linux go build -o main cmd/main.go | ||
|
|
||
|
|
||
| FROM ubuntu:18.04 |
There was a problem hiding this comment.
The base image ubuntu:18.04 reached its end of standard support in April 2023. Using an outdated base image can expose the application to unpatched security vulnerabilities. It is highly recommended to upgrade to a more recent and supported LTS version, such as ubuntu:22.04, or consider a smaller, more secure base image like alpine.
FROM ubuntu:22.04
| p := influxdb2.NewPoint(d.Influxdb2DataConfig.Measurement, | ||
| d.Influxdb2DataConfig.Tag, | ||
| map[string]interface{}{d.Influxdb2DataConfig.FieldKey: data.Value}, | ||
| time.Now()) |
There was a problem hiding this comment.
The AddData function uses time.Now() to timestamp the data point. However, the common.DataModel already contains a TimeStamp field (in milliseconds), which is set when the data is collected. Using time.Now() introduces a delay and uses a different timestamp from the actual data collection event. It's better to use the timestamp from the DataModel for data consistency.
| time.Now()) | |
| time.UnixMilli(data.TimeStamp)) |
| var ( | ||
| RedisCli *redis.Client | ||
| ) |
There was a problem hiding this comment.
Using a global variable RedisCli for the Redis client is not thread-safe and is considered bad practice. It can lead to race conditions and makes the code difficult to test. The Redis client should be a field within the DataBaseConfig struct, ensuring that each instance manages its own client.
var (
)| insertSQL := fmt.Sprintf("INSERT INTO %s USING %s TAGS ('%s') VALUES('%v','%s', '%s', '%s', '%s');", | ||
| legalTag, legalTable, legalTag, datatime, tableName, data.PropertyName, data.Value, data.Type) | ||
|
|
||
| rows, _ := DB.Query(stableName) |
There was a problem hiding this comment.
The error returned by DB.Query is being ignored. This is a bad practice as it can hide important issues, such as problems with the database connection or the query itself. The error should always be checked and handled appropriately.
rows, err := DB.Query(stableName)
if err != nil {
klog.Errorf("failed to query stable: %v", err)
return err
}| payload := data.PropertyName + "=" + data.Value | ||
| formatTimeStr := time.Unix(data.TimeStamp/1e3, 0).Format("2006-01-02 15:04:05") | ||
| currentTime := "&time" + "=" + formatTimeStr | ||
| payload += currentTime |
There was a problem hiding this comment.
The payload for an application/x-www-form-urlencoded request is constructed by simple string concatenation. This is unsafe, as special characters in data.PropertyName or data.Value (e.g., &, =) will break the payload format. Use the net/url package to properly encode the values.
form := url.Values{}
form.Add(data.PropertyName, data.Value)
form.Add("time", time.Unix(data.TimeStamp/1e3, 0).Format("2006-01-02 15:04:05"))
payload := form.Encode()| protocol: opcua # TODO add your protocol name | ||
| address: 127.0.0.1:1234 # TODO add your protocol address |
There was a problem hiding this comment.
The configuration file contains TODO comments, which suggests that the configuration might be incomplete or using placeholder values. These should be removed and replaced with appropriate default values before merging to avoid confusion and ensure the mapper is configured correctly out-of-the-box.
protocol: opcua
address: opc.tcp://127.0.0.1:4840 # Default OPC UA server address| @@ -0,0 +1,157 @@ | |||
| package tdengine | |||
There was a problem hiding this comment.
This file starts with a UTF-8 Byte Order Mark (BOM), which is unnecessary in Go source files and can sometimes cause issues with tools. It should be removed.
| package tdengine | |
| package tdengine |
2 participants
What type of PR is this?
/kind feature
What this PR does / why we need it:
This PR adds a built-in OPC UA mapper based on the v1beta1 version of the device management API and the new mapper-framework.
After the upgrade of the device management API to v1beta1, the existing built-in mappers based on v1alpha2 can no longer run in a KubeEdge cluster.
This change helps by extending built-in mapper support and improving the user experience for users migrating to v1beta1.
Which issue(s) this PR fixes:
Fixes kubeedge/kubeedge#6423