feat(libooniengine): add support for running tasks

internal/engine/session.go

@@ -221,6 +221,63 @@ func NewSession(ctx context.Context, config SessionConfig) (*Session, error) {
 	return sess, nil
 }
 
+func NewSessionWithoutTunnel(ctx context.Context, config *SessionConfig) (*Session, error) {
+	if config.Logger == nil {
+		return nil, errors.New("Logger is empty")
+	}
+	if config.SoftwareName == "" {
+		return nil, errors.New("SoftwareName is empty")
+	}
+	if config.SoftwareVersion == "" {
+		return nil, errors.New("SoftwareVersion is empty")
+	}
+	if config.KVStore == nil {
+		config.KVStore = &kvstore.Memory{}
+	}
+	// Implementation note: if config.TempDir is empty, then Go will
+	// use the temporary directory on the current system. This should
+	// work on Desktop. We tested that it did also work on iOS, but
+	// we have also seen on 2020-06-10 that it does not work on Android.
+	tempDir, err := ioutil.TempDir(config.TempDir, "ooniengine")
+	if err != nil {
+		return nil, err
+	}
+	config.Logger.Infof(
+		"ooniprobe-engine/v%s %s dirty=%s %s",
+		version.Version,
+		runtimex.BuildInfo.VcsRevision,
+		runtimex.BuildInfo.VcsModified,
+		runtimex.BuildInfo.GoVersion,
+	)
+	sess := &Session{
+		availableProbeServices:  config.AvailableProbeServices,
+		byteCounter:             bytecounter.New(),
+		kvStore:                 config.KVStore,
+		logger:                  config.Logger,
+		queryProbeServicesCount: &atomic.Int64{},
+		softwareName:            config.SoftwareName,
+		softwareVersion:         config.SoftwareVersion,
+		tempDir:                 tempDir,
+		torArgs:                 config.TorArgs,
+		torBinary:               config.TorBinary,
+		tunnelDir:               config.TunnelDir,
+	}
+	var proxyURL *url.URL = nil
+	sess.proxyURL = proxyURL
+	sess.resolver = &sessionresolver.Resolver{
+		ByteCounter: sess.byteCounter,
+		KVStore:     config.KVStore,
+		Logger:      sess.logger,
+		ProxyURL:    proxyURL,
+	}
+	txp := netxlite.NewHTTPTransportWithLoggerResolverAndOptionalProxyURL(
+		sess.logger, sess.resolver, sess.proxyURL,
+	)
+	txp = bytecounter.WrapHTTPTransport(txp, sess.byteCounter)
+	sess.httpDefaultTransport = txp
+	return sess, nil
+}
+
 // TunnelDir returns the persistent directory used by tunnels.
 func (s *Session) TunnelDir() string {
 	return s.tunnelDir
@@ -393,6 +450,10 @@ func (s *Session) MaybeLookupBackends() error {
 	return s.MaybeLookupBackendsContext(context.Background())
 }
 
+func (s *Session) Resolver() *sessionresolver.Resolver {
+	return s.resolver
+}
+
 // ErrAlreadyUsingProxy indicates that we cannot create a tunnel with
 // a specific name because we already configured a proxy.
 var ErrAlreadyUsingProxy = errors.New(

internal/libooniengine/emitter.go

@@ -0,0 +1,21 @@
+package main
+
+//
+// Emitter
+//
+
+// taskEmitter implements taskMaybeEmitter.
+type taskChanEmitter struct {
+	// out is the channel where we emit events.
+	out chan *response
+}
+
+var _ taskMaybeEmitter = &taskChanEmitter{}
+
+// maybeEmitEvent implements taskMaybeEmitter.maybeEmitEvent.
+func (e *taskChanEmitter) maybeEmitEvent(resp *response) {
+	select {
+	case e.out <- resp:
+	default: // buffer full, discard this event
+	}
+}

internal/libooniengine/engine.h

@@ -7,6 +7,13 @@
 /// C API for using the OONI engine.
 ///
 
+#include <stdint.h>
+
+/// OONITask is an asynchronous thread of execution managed by the OONI
+/// engine that performs a background operation and emits meaningful
+/// events such as, for example, the results of measurements.
+typedef uintptr_t OONITask;
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -19,7 +26,53 @@ char *OONIEngineVersion(void);
 /// OONIEngineFreeMemory frees the memory allocated by the engine.
 ///
 /// @param ptr a void pointer refering to the memory to be freed.
-void OONIENgineFreeMemory(void *ptr);
+void OONIEngineFreeMemory(void *ptr);
+
+/// OONIEngineCall starts a new OONITask using the given [req].
+///
+/// @param req A JSON string, owned by the caller, that
+/// contains the configuration for the task to start.  
+///
+/// @return Zero on failure, nonzero on success. If the return value
+/// is nonzero, a task is running. In such a case, the caller is
+/// responsible to eventually dispose of the task using OONIEngineFreeMemory.
+OONITask OONIEngineCall(char *req);
+
+/// OONIEngineWaitForNextEvent awaits on the [task] event queue until
+/// a new event is available or the given [timeout] expires.
+///
+/// @param task Task handle returned by OONIEngineCall.
+///
+/// @param timeout Timeout in milliseconds. If the timeout is zero
+/// or negative, this function would potentially block forever.
+///
+/// @return A NULL pointer on failure, non-NULL otherwise. If the return
+/// value is non-NULL, the caller takes ownership of the OONIMessage
+/// pointer and MUST free it using OONIMessageFree when done using it.
+///
+/// This function will return NULL:
+///
+/// 1. when the timeout expires;
+///
+/// 2. if [task] is done;
+///
+/// 3. if [task] is zero or does not refer to a valid task;
+///
+/// 4. if we cannot protobuf serialize the message;
+///
+/// 5. possibly because of other unknown internal errors.
+///
+/// In short, you cannot reliably determine whether a task is done by
+/// checking whether this function has returned NULL.
+char *OONIEngineWaitForNextEvent(OONITask task, int32_t timeout);
+
+/// OONIEngineInterrupt tells [task] to stop ASAP.
+///
+/// @param task Task handle returned by OONIEngineCall. If task is zero
+/// or does not refer to a valid task, this function will just do nothing.
+void OONIEngineInterrupt(OONITask task);
+
+// OONITask
 
 #ifdef __cplusplus
 }

internal/libooniengine/geolocate.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"context"
+
+	"github.com/ooni/probe-cli/v3/internal/geolocate"
+)
+
+func init() {
+	taskRegistry["Geolocate"] = &geolocateTaskRunner{}
+}
+
+// geolocateOptions contains the request arguments for the Geolocate task.
+type geolocateOptions struct {
+	SessionId int64 `json:",omitempty"`
+}
+
+// geolocateResponse is the response for the Geolocate task.
+type geolocateResponse struct {
+	ASN                 uint   `json:",omitempty"`
+	CountryCode         string `json:",omitempty"`
+	NetworkName         string `json:",omitempty"`
+	ProbeIP             string `json:",omitempty"`
+	ResolverASN         uint   `json:",omitempty"`
+	ResolverIP          string `json:",omitempty"`
+	ResolverNetworkName string `json:",omitempty"`
+	Error               string `json:",omitempty"`
+}
+
+type geolocateTaskRunner struct{}
+
+var _ taskRunner = &geolocateTaskRunner{}
+
+// main implements taskRunner.main
+func (tr *geolocateTaskRunner) main(ctx context.Context,
+	emitter taskMaybeEmitter, req *request, resp *response) {
+	logger := newTaskLogger(emitter)
+	sessionId := req.Geolocate.SessionId
+	sess := mapSession[sessionId]
+	if sess == nil {
+		logger.Warnf("session: %s", errInvalidSessionId.Error())
+		resp.Geolocate.Error = errInvalidSessionId.Error()
+		return
+	}
+	gt := geolocate.NewTask(geolocate.Config{
+		Logger:    sess.Logger(),
+		Resolver:  sess.Resolver(),
+		UserAgent: sess.UserAgent(),
+	})
+	results, err := gt.Run(ctx)
+	if err != nil {
+		logger.Warnf("geolocate: %s", err.Error())
+		resp.Geolocate.Error = err.Error()
+		return
+	}
+	resp = &response{
+		Geolocate: geolocateResponse{
+			ASN:                 results.ASN,
+			CountryCode:         results.CountryCode,
+			NetworkName:         results.NetworkName,
+			ProbeIP:             results.ProbeIP,
+			ResolverASN:         results.ResolverASN,
+			ResolverIP:          results.ResolverIP,
+			ResolverNetworkName: results.ResolverNetworkName,
+		},
+	}
+}

internal/libooniengine/logger.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"fmt"
+
+	"github.com/ooni/probe-cli/v3/internal/model"
+)
+
+type LogLevel int32
+
+const (
+	// The DEBUG log level.
+	logLevel_DEBUG LogLevel = 0
+	// The INFO log level.
+	logLevel_INFO LogLevel = 1
+	// The WARNING log level.
+	logLevel_WARNING LogLevel = 2
+)
+
+type logResponse struct {
+	Level   LogLevel `json:",omitempty"`
+	Message string   `json:",omitempty"`
+}
+
+// taskLogger implements model.Logger for tasks.
+type taskLogger struct {
+	// emitter is used to emit log events.
+	emitter taskMaybeEmitter
+
+	// verbose indicates whether verbose logging is enabled.
+	verbose bool
+}
+
+// newLogger creates a new taskLogger instance using
+// the [emitter] to emit log events.
+func newTaskLogger(emitter taskMaybeEmitter) *taskLogger {
+	return &taskLogger{
+		emitter: emitter,
+		verbose: false,
+	}
+}
+
+var _ model.Logger = &taskLogger{}
+
+// Debugf implements model.Logger.Debugf.
+func (tl *taskLogger) Debugf(format string, values ...any) {
+	if tl.verbose {
+		tl.emit(logLevel_DEBUG, fmt.Sprintf(format, values...))
+	}
+}
+
+// Debug implements model.Logger.Debug.
+func (tl *taskLogger) Debug(message string) {
+	if tl.verbose {
+		tl.emit(logLevel_DEBUG, message)
+	}
+}
+
+// Infof implements model.Logger.Infof.
+func (tl *taskLogger) Infof(format string, values ...any) {
+	tl.emit(logLevel_INFO, fmt.Sprintf(format, values...))
+}
+
+// Info implements model.Logger.Info.
+func (tl *taskLogger) Info(message string) {
+	tl.emit(logLevel_INFO, message)
+}
+
+// Warnf implements model.Logger.Warnf.
+func (tl *taskLogger) Warnf(format string, values ...any) {
+	tl.emit(logLevel_WARNING, fmt.Sprintf(format, values...))
+}
+
+// Warn implements model.Logger.Warn.
+func (tl *taskLogger) Warn(message string) {
+	tl.emit(logLevel_WARNING, message)
+}
+
+// emit emits a log message.
+func (tl *taskLogger) emit(level LogLevel, message string) {
+	logResp := &response{
+		Logger: logResponse{
+			Level:   level,
+			Message: message,
+		},
+	}
+	tl.emitter.maybeEmitEvent(logResp)
+}