thmeitz / ksqldb-go Goto Github PK

push.go

Hi,

I am very new with go and I don't know how to access the values of the headerChannel from the push api example.

go func() {
		for row := range rowChannel {
			if row != nil {
				for row_col := range row {
					fieldn := row[row_col].(string)
					//var header ksqldb.Header
					header := <- headerChannel
					fmt.Printf("%+v\n",header)
					fmt.Printf("%d : %s\n", row_col, fieldn)
				}
				
				
			}
		}
		cancelFunction()
	}()

I did something like this in the go routine. The header values are displayed in the console :
{queryId:transient_USER_BEHAVIOR_4058313251310728391 columns:[{Name:EXE Type:STRING}]}

But I cannot access any field of the Header type. I have the following error header.queryId undefined (type ksqldb.Header has no field or method if I call this instruction :
fmt.Printf("%s\n",header.queryId)

Do you know what is the problem ?

I use go 1.18 with ksqldb-go v0.0.4

Currently, there is no way to propagate a context to HTTP requests which means that the client does not respect cancellations and timeouts and tracing is also not possible.

Generally, context support can be added in a backwards compatible way by creating new functions for each request, attaching Context or Ctx as a suffix.

Would you be open to accepting a contribution with this change?

(e.g. commandID, commandStatus.status, etc).

Hello,

It looks like SQL injection is still possible on ksqldb package if the parameter has single quotes in it.

k := `SELECT TIMESTAMPTOSTRING(WINDOWSTART,'yyyy-MM-dd HH:mm:ss','Europe/London') AS WINDOW_START,
TIMESTAMPTOSTRING(WINDOWEND,'HH:mm:ss','Europe/London') AS WINDOW_END,
DOG_SIZE, DOGS_CT FROM DOGS_BY_SIZE
WHERE DOG_SIZE=?;`

stmnt, err := ksqldb.QueryBuilder(k, "middle' OR 'a'='a")
if err != nil {
	log.Fatal(err)
}
fmt.Println(*stmnt)

See: https://go.dev/play/p/4KoWROjK903

The Cobra example results in too many dependencies being added to the package. For this reason, the examples are moved to the ksqldb-go-examples repo.

confluentinc/ksql@db76b3e#diff-3c7c08413bb34acade8053b6c1b720d042e446a3996cd1144bf3a895a5ef276b

We ran into SQL syntax problems with executing queries when the raw string literal doesn't contain spaces after \n.

eg.:

q := `CREATE STREAM IF NOT EXISTS DOGS (ID STRING KEY,
	NAME STRING,
	DOGSIZE STRING,
	AGE STRING)
WITH (KAFKA_TOPIC='dogs',
VALUE_FORMAT='JSON', PARTITIONS=1);
`

instead of:

q := `CREATE STREAM IF NOT EXISTS DOGS (ID STRING KEY, 
	NAME STRING, 
	DOGSIZE STRING, 
	AGE STRING) 
WITH (KAFKA_TOPIC='dogs', 
VALUE_FORMAT='JSON', PARTITIONS=1);
`

I think it's very error-prone, and the SanitizeQuery function should replace \t and \n with .

implement close-push-query

https://docs.ksqldb.io/en/latest/developer-guide/ksqldb-rest-api/streaming-endpoint/#terminating-queries

InsecureSkipVerify

SQL strings should be built by a QueryBuilder. Otherwise the system is open for SQL injections (see https://github.com/OWASP/Go-SCP/blob/master/dist/go-webapp-scp.pdf).

See https://sonarcloud.io/project/issues?id=thmeitz_ksqldb-go&issues=AXznwv_A0Y5mJN1ndwx9&open=AXznwv_A0Y5mJN1ndwx9

// it's a kind of Prepared Statement
// its an example from go-webpp-scp.pdf ;)
query := "SELECT number, expireDate, cvv FROM creditcards WHERE customerId = ?"
stmt, err := ksqldb.QueryBuilder(ctx, query).Int(customerId)

Types:

• all types ksqldb supports

Errors:

• type errors (can't convert customerId to Int, etc)
• to few/many parameters

Thoughts:

• do we need Context?

This fixes this bug

• examples/cobra-test/cmd/pull.go

• use DialTLSContext instead

https://docs.ksqldb.io/en/latest/developer-guide/api/index.html#content-types

• Execute a statement (/ksql endpoint)
• Run push and pull queries (/query-stream endpoint)
• Terminate a cluster (/ksql/terminate endpoint)
• Introspect query status (/status endpoint)
• Introspect server status (/info endpoint)
• Introspect cluster status (/clusterStatus endpoint)
• Get the validity of a property (/is_valid_property)

Run a query (/query endpoint) is deprecated and will not be implemented

• write tests for push
• write tests for pull
• refactor duplications
• push - change method signature to support QueryOptions