diff --git a/core/core.go b/core/core.go index 8b70cba6..42645f0f 100644 --- a/core/core.go +++ b/core/core.go @@ -7,11 +7,12 @@ import ( "github.com/safing/portbase/modules" "github.com/safing/portbase/modules/subsystems" + "github.com/safing/portmaster/updates" _ "github.com/safing/portmaster/broadcasts" _ "github.com/safing/portmaster/netenv" + _ "github.com/safing/portmaster/netquery" _ "github.com/safing/portmaster/status" _ "github.com/safing/portmaster/ui" - "github.com/safing/portmaster/updates" ) const ( @@ -26,7 +27,7 @@ var ( ) func init() { - module = modules.Register("core", prep, start, nil, "base", "subsystems", "status", "updates", "api", "notifications", "ui", "netenv", "network", "interception", "compat", "broadcasts") + module = modules.Register("core", prep, start, nil, "base", "subsystems", "status", "updates", "api", "notifications", "ui", "netenv", "network", "netquery", "interception", "compat", "broadcasts") subsystems.Register( "core", "Core", diff --git a/go.mod b/go.mod index c46806ed..95a6a374 100644 --- a/go.mod +++ b/go.mod @@ -18,6 +18,14 @@ require ( github.com/safing/spn v0.4.12 github.com/shirou/gopsutil v3.21.11+incompatible github.com/spf13/cobra v1.5.0 + github.com/stretchr/testify v1.7.1 + github.com/tannerryan/ring v1.1.2 + github.com/tevino/abool v1.2.0 + github.com/umahmood/haversine v0.0.0-20151105152445-808ab04add26 + golang.org/x/net v0.0.0-20220708220712-1185a9018129 + golang.org/x/sync v0.0.0-20220601150217-0de741cfad7f + golang.org/x/sys v0.0.0-20220712014510-0a85c31ab51e + zombiezen.com/go/sqlite v0.10.0 github.com/stretchr/testify v1.8.0 github.com/tannerryan/ring v1.1.2 github.com/tevino/abool v1.2.0 @@ -39,12 +47,14 @@ require ( github.com/go-ole/go-ole v1.2.6 // indirect github.com/gofrs/uuid v4.2.0+incompatible // indirect github.com/google/go-cmp v0.5.8 // indirect + github.com/google/uuid v1.3.0 // indirect github.com/gorilla/mux v1.8.0 // indirect github.com/gorilla/websocket v1.5.0 // indirect github.com/hashicorp/errwrap v1.1.0 // indirect github.com/inconshreveable/mousetrap v1.0.0 // indirect github.com/josharian/native v1.0.0 // indirect github.com/kardianos/osext v0.0.0-20190222173326-2bc1f35cddc0 // indirect + github.com/mattn/go-isatty v0.0.14 // indirect github.com/mdlayher/netlink v1.6.0 // indirect github.com/mdlayher/socket v0.2.3 // indirect github.com/mitchellh/copystructure v1.2.0 // indirect @@ -52,6 +62,7 @@ require ( github.com/mr-tron/base58 v1.2.0 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect github.com/rot256/pblind v0.0.0-20211117203330-22455f90b565 // indirect github.com/safing/jess v0.2.3 // indirect github.com/satori/go.uuid v1.2.0 // indirect @@ -75,5 +86,9 @@ require ( golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4 // indirect golang.org/x/tools v0.1.11 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect + gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect + modernc.org/libc v1.16.17 // indirect + modernc.org/mathutil v1.4.1 // indirect + modernc.org/memory v1.1.1 // indirect + modernc.org/sqlite v1.17.3 // indirect ) diff --git a/go.sum b/go.sum index 0ba1303f..168ff329 100644 --- a/go.sum +++ b/go.sum @@ -66,6 +66,7 @@ cloud.google.com/go/trace v0.1.0/go.mod h1:wxEwsoeRVPbeSkt7ZC9nWCgmoKQRAoySN7XHW contrib.go.opencensus.io/exporter/aws v0.0.0-20200617204711-c478e41e60e9/go.mod h1:uu1P0UCM/6RbsMrgPa98ll8ZcHM858i/AD06a9aLRCA= contrib.go.opencensus.io/exporter/stackdriver v0.13.8/go.mod h1:huNtlWx75MwO7qMs0KrMxPZXzNNWebav1Sq/pm02JdQ= contrib.go.opencensus.io/integrations/ocsql v0.1.7/go.mod h1:8DsSdjz3F+APR+0z0WkU1aRorQCFfRxvqjUUPMbF3fE= +crawshaw.io/iox v0.0.0-20181124134642-c51c3df30797/go.mod h1:sXBiorCo8c46JlQV3oXPKINnZ8mcqnye1EkVkqsectk= dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU= github.com/AlecAivazis/survey/v2 v2.0.7/go.mod h1:mlizQTaPjnR4jcpwRSaSlkbsRfYFEyKgLQvYTzxxiHA= github.com/AndreasBriese/bbloom v0.0.0-20190825152654-46b345b51c96 h1:cTp8I5+VIoKjsnZuH8vjyaysT/ses3EvZeaV/1UkF2M= @@ -460,6 +461,7 @@ github.com/google/subcommands v1.0.1/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3 github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/wire v0.5.0/go.mod h1:ngWDr9Qvq3yZA10YrxfyGELY/AFWGVpy9c1LTRi1EoU= github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+vpHVxEJEs9eg= @@ -649,7 +651,9 @@ github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hd github.com/mattn/go-isatty v0.0.10/go.mod h1:qgIWMr58cqv1PHHyhnkY9lrL7etaEgOFcMEpPG5Rm84= github.com/mattn/go-isatty v0.0.11/go.mod h1:PhnuNfih5lzO57/f3n+odYbM4JtupLOxQOAqxQCu2WE= github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU= +github.com/mattn/go-isatty v0.0.14 h1:yVuAays6BHfxijgZPzw+3Zlu5yQgKGP2/hcQbHb7S9Y= github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94= +github.com/mattn/go-sqlite3 v1.14.12/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU= github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= github.com/mdlayher/ethtool v0.0.0-20210210192532-2b88debcdd43/go.mod h1:+t7E0lkKfbBsebllff1xdTmyJt8lH37niI6kwFk9OTo= github.com/mdlayher/ethtool v0.0.0-20211028163843-288d040e9d60/go.mod h1:aYbhishWc4Ai3I2U4Gaa2n3kHWSwzme6EsG/46HRQbE= @@ -792,6 +796,8 @@ github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7z github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA= github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A= github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU= +github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 h1:OdAsTTz6OkFY5QxjkYwrChwuRruF69c169dPK26NUlk= +github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6SoW27p1b0cqNHllgS5HIMJraePCO15w5zCzIWYg= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= @@ -1224,8 +1230,6 @@ golang.org/x/net v0.0.0-20220412020605-290c469a71a5/go.mod h1:CfG3xpIq0wQ8r1q4Su golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/net v0.0.0-20220513224357-95641704303c/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/net v0.0.0-20220621193019-9d032be2e588/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= -golang.org/x/net v0.0.0-20220622184535-263ec571b305 h1:dAgbJ2SP4jD6XYfMNLVj0BF21jo2PjChrtGaAvF5M3I= -golang.org/x/net v0.0.0-20220622184535-263ec571b305/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -1387,8 +1391,6 @@ golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220513210249-45d2b4557a2a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220615213510-4f61da869c0c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664 h1:wEZYwx+kK+KlZ0hpvP2Ls1Xr4+RWnlzGFwPP0aiDjIU= -golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= @@ -1468,6 +1470,7 @@ golang.org/x/tools v0.0.0-20200808161706-5bf02b21f123/go.mod h1:njjCfa9FT2d7l9Bc golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA= golang.org/x/tools v0.0.0-20200904185747-39188db58858/go.mod h1:Cj7w3i3Rnn0Xh82ur9kSqwfTHTeVxaDqrfMjpcNT6bE= golang.org/x/tools v0.0.0-20201110124207-079ba7bd75cd/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/tools v0.0.0-20201124115921-2c860bdd6e78/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.0.0-20201201161351-ac6f37ff4c2a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.0.0-20201208233053-a543418bbed2/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= golang.org/x/tools v0.0.0-20210105154028-b0ab187a4818/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= @@ -1710,6 +1713,32 @@ k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE= k8s.io/klog/v2 v2.9.0/go.mod h1:hy9LJ/NvuK+iVyP4Ehqva4HxZG/oXyIS3n3Jmire4Ec= k8s.io/kube-openapi v0.0.0-20210421082810-95288971da7e/go.mod h1:vHXdDvt9+2spS2Rx9ql3I8tycm3H9FDfdUoIuKCefvw= k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA= +lukechampine.com/uint128 v1.1.1/go.mod h1:c4eWIwlEGaxC/+H1VguhU4PHXNWDCDMUlWdIWl2j1gk= +modernc.org/cc/v3 v3.36.0/go.mod h1:NFUHyPn4ekoC/JHeZFfZurN6ixxawE1BnVonP/oahEI= +modernc.org/ccgo/v3 v3.0.0-20220428102840-41399a37e894/go.mod h1:eI31LL8EwEBKPpNpA4bU1/i+sKOwOrQy8D87zWUcRZc= +modernc.org/ccgo/v3 v3.0.0-20220430103911-bc99d88307be/go.mod h1:bwdAnOoaIt8Ax9YdWGjxWsdkPcZyRPHqrOvJxaKAKGw= +modernc.org/ccgo/v3 v3.16.4/go.mod h1:tGtX0gE9Jn7hdZFeU88slbTh1UtCYKusWOoCJuvkWsQ= +modernc.org/ccgo/v3 v3.16.6/go.mod h1:tGtX0gE9Jn7hdZFeU88slbTh1UtCYKusWOoCJuvkWsQ= +modernc.org/ccorpus v1.11.6/go.mod h1:2gEUTrWqdpH2pXsmTM1ZkjeSrUWDpjMu2T6m29L/ErQ= +modernc.org/httpfs v1.0.6/go.mod h1:7dosgurJGp0sPaRanU53W4xZYKh14wfzX420oZADeHM= +modernc.org/libc v0.0.0-20220428101251-2d5f3daf273b/go.mod h1:p7Mg4+koNjc8jkqwcoFBJx7tXkpj00G77X7A72jXPXA= +modernc.org/libc v1.16.0/go.mod h1:N4LD6DBE9cf+Dzf9buBlzVJndKr/iJHG97vGLHYnb5A= +modernc.org/libc v1.16.1/go.mod h1:JjJE0eu4yeK7tab2n4S1w8tlWd9MxXLRzheaRnAKymU= +modernc.org/libc v1.16.7/go.mod h1:hYIV5VZczAmGZAnG15Vdngn5HSF5cSkbvfz2B7GRuVU= +modernc.org/libc v1.16.17 h1:rXo8IZJvP+QSN1KrlV23dtkM3XfGYXjx3RbLLzBtndM= +modernc.org/libc v1.16.17/go.mod h1:hYIV5VZczAmGZAnG15Vdngn5HSF5cSkbvfz2B7GRuVU= +modernc.org/mathutil v1.2.2/go.mod h1:mZW8CKdRPY1v87qxC/wUdX5O1qDzXMP5TH3wjfpga6E= +modernc.org/mathutil v1.4.1 h1:ij3fYGe8zBF4Vu+g0oT7mB06r8sqGWKuJu1yXeR4by8= +modernc.org/mathutil v1.4.1/go.mod h1:mZW8CKdRPY1v87qxC/wUdX5O1qDzXMP5TH3wjfpga6E= +modernc.org/memory v1.1.1 h1:bDOL0DIDLQv7bWhP3gMvIrnoFw+Eo6F7a2QK9HPDiFU= +modernc.org/memory v1.1.1/go.mod h1:/0wo5ibyrQiaoUoH7f9D8dnglAmILJ5/cxZlRECf+Nw= +modernc.org/opt v0.1.1/go.mod h1:WdSiB5evDcignE70guQKxYUl14mgWtbClRi5wmkkTX0= +modernc.org/sqlite v1.17.3 h1:iE+coC5g17LtByDYDWKpR6m2Z9022YrSh3bumwOnIrI= +modernc.org/sqlite v1.17.3/go.mod h1:10hPVYar9C0kfXuTWGz8s0XtB8uAGymUy51ZzStYe3k= +modernc.org/strutil v1.1.1/go.mod h1:DE+MQQ/hjKBZS2zNInV5hhcipt5rLPWkmpbGeW5mmdw= +modernc.org/tcl v1.13.1/go.mod h1:XOLfOwzhkljL4itZkK6T72ckMgvj0BDsnKNdZVUOecw= +modernc.org/token v1.0.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= +modernc.org/z v1.5.1/go.mod h1:eWFB510QWW5Th9YGZT81s+LwvaAs3Q2yr4sP0rmLkv8= nhooyr.io/websocket v1.8.6/go.mod h1:B70DZP8IakI65RVQ51MsWP/8jndNma26DVA/nFSCgW0= nhooyr.io/websocket v1.8.7/go.mod h1:B70DZP8IakI65RVQ51MsWP/8jndNma26DVA/nFSCgW0= rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8= @@ -1719,3 +1748,5 @@ rsc.io/sampler v1.3.0/go.mod h1:T1hPZKmBbMNahiBKFy5HrXp6adAjACjK9JXDnKaTXpA= sigs.k8s.io/structured-merge-diff/v4 v4.0.2/go.mod h1:bJZC9H9iH24zzfZ/41RGcq60oK1F7G282QMXDPYydCw= sigs.k8s.io/structured-merge-diff/v4 v4.1.2/go.mod h1:j/nl6xW8vLS49O8YvXW1ocPhZawJtm+Yrr7PPRQ0Vg4= sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc= +zombiezen.com/go/sqlite v0.10.0 h1:hegW0Y8c/fSJ2VjbjBeiKJaQOISNr4EUTx1VZx94Q9Y= +zombiezen.com/go/sqlite v0.10.0/go.mod h1:tOd9u3peffVYnXOedepSJmX92n/mbqf594wcJ+29jf8= diff --git a/netquery/chart_handler.go b/netquery/chart_handler.go new file mode 100644 index 00000000..aaff4892 --- /dev/null +++ b/netquery/chart_handler.go @@ -0,0 +1,140 @@ +package netquery + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "io/ioutil" + "net/http" + "strings" + + "github.com/safing/portmaster/netquery/orm" +) + +type ChartHandler struct { + Database *Database +} + +func (ch *ChartHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { + requestPayload, err := ch.parseRequest(req) + if err != nil { + http.Error(resp, err.Error(), http.StatusBadRequest) + + return + } + + query, paramMap, err := requestPayload.generateSQL(req.Context(), ch.Database.Schema) + if err != nil { + http.Error(resp, err.Error(), http.StatusBadRequest) + + return + } + + // actually execute the query against the database and collect the result + var result []map[string]interface{} + if err := ch.Database.Execute( + req.Context(), + query, + orm.WithNamedArgs(paramMap), + orm.WithResult(&result), + orm.WithSchema(*ch.Database.Schema), + ); err != nil { + http.Error(resp, "Failed to execute query: "+err.Error(), http.StatusInternalServerError) + + return + } + + // send the HTTP status code + resp.WriteHeader(http.StatusOK) + + // prepare the result encoder. + enc := json.NewEncoder(resp) + enc.SetEscapeHTML(false) + enc.SetIndent("", " ") + + enc.Encode(map[string]interface{}{ + "results": result, + "query": query, + "params": paramMap, + }) +} + +func (ch *ChartHandler) parseRequest(req *http.Request) (*QueryActiveConnectionChartPayload, error) { + var body io.Reader + + switch req.Method { + case http.MethodPost, http.MethodPut: + body = req.Body + case http.MethodGet: + body = strings.NewReader(req.URL.Query().Get("q")) + default: + return nil, fmt.Errorf("invalid HTTP method") + } + + var requestPayload QueryActiveConnectionChartPayload + blob, err := ioutil.ReadAll(body) + if err != nil { + return nil, fmt.Errorf("failed to read body" + err.Error()) + } + + body = bytes.NewReader(blob) + + dec := json.NewDecoder(body) + dec.DisallowUnknownFields() + + if err := json.Unmarshal(blob, &requestPayload); err != nil && !errors.Is(err, io.EOF) { + return nil, fmt.Errorf("invalid query: %w", err) + } + + return &requestPayload, nil +} + +func (req *QueryActiveConnectionChartPayload) generateSQL(ctx context.Context, schema *orm.TableSchema) (string, map[string]interface{}, error) { + template := ` +WITH RECURSIVE epoch(x) AS ( + SELECT strftime('%%s')-600 + UNION ALL + SELECT x+1 FROM epoch WHERE x+1 < strftime('%%s')+0 +) +SELECT x as timestamp, SUM(verdict IN (2, 5, 6)) AS value, SUM(verdict NOT IN (2, 5, 6)) as countBlocked FROM epoch + JOIN connections + ON strftime('%%s', connections.started)+0 <= timestamp+0 AND (connections.ended IS NULL OR strftime('%%s', connections.ended)+0 >= timestamp+0) + %s + GROUP BY round(timestamp/10, 0)*10;` + + clause, params, err := req.Query.toSQLWhereClause(ctx, "", schema, orm.DefaultEncodeConfig) + if err != nil { + return "", nil, err + } + + if params == nil { + params = make(map[string]interface{}) + } + + if req.TextSearch != nil { + textSearch, textParams, err := req.TextSearch.toSQLConditionClause(ctx, schema, "", orm.DefaultEncodeConfig) + if err != nil { + return "", nil, err + } + + if textSearch != "" { + if clause != "" { + clause += " AND " + } + clause += textSearch + + for key, val := range textParams { + params[key] = val + } + } + } + + if clause == "" { + return fmt.Sprintf(template, ""), map[string]interface{}{}, nil + } + + return fmt.Sprintf(template, "WHERE ( "+clause+")"), params, nil +} diff --git a/netquery/database.go b/netquery/database.go new file mode 100644 index 00000000..57aa6f6c --- /dev/null +++ b/netquery/database.go @@ -0,0 +1,315 @@ +package netquery + +import ( + "context" + "encoding/json" + "fmt" + "io" + "strings" + "sync" + "time" + + "github.com/safing/portbase/log" + "github.com/safing/portmaster/netquery/orm" + "github.com/safing/portmaster/network" + "github.com/safing/portmaster/network/netutils" + "github.com/safing/portmaster/network/packet" + "zombiezen.com/go/sqlite" + "zombiezen.com/go/sqlite/sqlitex" +) + +// InMemory is the "file path" to open a new in-memory database. +const InMemory = ":memory:" + +// Available connection types as their string representation. +const ( + ConnTypeDNS = "dns" + ConnTypeIP = "ip" +) + +// ConnectionTypeToString is a lookup map to get the string representation +// of a network.ConnectionType as used by this package. +var ConnectionTypeToString = map[network.ConnectionType]string{ + network.DNSRequest: ConnTypeDNS, + network.IPConnection: ConnTypeIP, +} + +type ( + // Database represents a SQLite3 backed connection database. + // It's use is tailored for persistance and querying of network.Connection. + // Access to the underlying SQLite database is synchronized. + // + // TODO(ppacher): somehow I'm receiving SIGBUS or SIGSEGV when no doing + // synchronization in *Database. Check what exactly sqlite.OpenFullMutex, etc.. + // are actually supposed to do. + // + Database struct { + Schema *orm.TableSchema + + l sync.Mutex + conn *sqlite.Conn + } + + // Conn is a network connection that is stored in a SQLite database and accepted + // by the *Database type of this package. This also defines, using the ./orm package, + // the table schema and the model that is exposed via the runtime database as well as + // the query API. + // + // Use ConvertConnection from this package to convert a network.Connection to this + // representation. + Conn struct { + // ID is a device-unique identifier for the connection. It is built + // from network.Connection by hashing the connection ID and the start + // time. We cannot just use the network.Connection.ID because it is only unique + // as long as the connection is still active and might be, although unlikely, + // reused afterwards. + ID string `sqlite:"id,primary"` + ProfileID string `sqlite:"profile"` + Path string `sqlite:"path"` + Type string `sqlite:"type,varchar(8)"` + External bool `sqlite:"external"` + IPVersion packet.IPVersion `sqlite:"ip_version"` + IPProtocol packet.IPProtocol `sqlite:"ip_protocol"` + LocalIP string `sqlite:"local_ip"` + LocalPort uint16 `sqlite:"local_port"` + RemoteIP string `sqlite:"remote_ip"` + RemotePort uint16 `sqlite:"remote_port"` + Domain string `sqlite:"domain"` + Country string `sqlite:"country,varchar(2)"` + ASN uint `sqlite:"asn"` + ASOwner string `sqlite:"as_owner"` + Latitude float64 `sqlite:"latitude"` + Longitude float64 `sqlite:"longitude"` + Scope netutils.IPScope `sqlite:"scope"` + Verdict network.Verdict `sqlite:"verdict"` + Started time.Time `sqlite:"started,text,time"` + Ended *time.Time `sqlite:"ended,text,time"` + Tunneled bool `sqlite:"tunneled"` + Encrypted bool `sqlite:"encrypted"` + Internal bool `sqlite:"internal"` + Direction string `sqlite:"direction"` + ExtraData json.RawMessage `sqlite:"extra_data"` + Allowed *bool `sqlite:"allowed"` + ProfileRevision int `sqlite:"profile_revision"` + ExitNode *string `sqlite:"exit_node"` + + // FIXME(ppacher): support "NOT" in search query to get rid of the following helper fields + SPNUsed bool `sqlite:"spn_used"` // could use "exit_node IS NOT NULL" or "exit IS NULL" + Active bool `sqlite:"active"` // could use "ended IS NOT NULL" or "ended IS NULL" + + // FIXME(ppacher): we need to profile here for "suggestion" support. It would be better to keep a table of profiles in sqlite and use joins here + ProfileName string `sqlite:"profile_name"` + } +) + +// New opens a new database at path. The database is opened with Full-Mutex, Write-Ahead-Log (WAL) +// and Shared-Cache enabled. +// +// TODO(ppacher): check which sqlite "open flags" provide the best performance and don't cause +// SIGBUS/SIGSEGV when used with out a dedicated mutex in *Database. +// +func New(path string) (*Database, error) { + c, err := sqlite.OpenConn( + path, + sqlite.OpenCreate, + sqlite.OpenReadWrite, + sqlite.OpenFullMutex, + sqlite.OpenWAL, + sqlite.OpenSharedCache, + ) + if err != nil { + return nil, fmt.Errorf("failed to open sqlite at %s: %w", path, err) + } + + schema, err := orm.GenerateTableSchema("connections", Conn{}) + if err != nil { + return nil, err + } + + return &Database{ + Schema: schema, + conn: c, + }, nil +} + +// NewInMemory is like New but creates a new in-memory database and +// automatically applies the connection table schema. +func NewInMemory() (*Database, error) { + db, err := New(InMemory) + if err != nil { + return nil, err + } + + // this should actually never happen because an in-memory database + // always starts empty... + if err := db.ApplyMigrations(); err != nil { + return nil, fmt.Errorf("failed to prepare database: %w", err) + } + + return db, nil +} + +// ApplyMigrations applies any table and data migrations that are needed +// to bring db up-to-date with the built-in schema. +// TODO(ppacher): right now this only applies the current schema and ignores +// any data-migrations. Once the history module is implemented this should +// become/use a full migration system -- use zombiezen.com/go/sqlite/sqlitemigration +func (db *Database) ApplyMigrations() error { + // get the create-table SQL statement from the infered schema + sql := db.Schema.CreateStatement(false) + + // execute the SQL + if err := sqlitex.ExecuteTransient(db.conn, sql, nil); err != nil { + return fmt.Errorf("failed to create schema: %w", err) + } + + return nil +} + +// Execute executes a custom SQL query against the SQLite database used by db. +// It uses orm.RunQuery() under the hood so please refer to the orm package for +// more information about available options. +func (db *Database) Execute(ctx context.Context, sql string, args ...orm.QueryOption) error { + db.l.Lock() + defer db.l.Unlock() + + return orm.RunQuery(ctx, db.conn, sql, args...) +} + +// CountRows returns the number of rows stored in the database. +func (db *Database) CountRows(ctx context.Context) (int, error) { + var result []struct { + Count int `sqlite:"count"` + } + + if err := db.Execute(ctx, "SELECT COUNT(*) AS count FROM connections", orm.WithResult(&result)); err != nil { + return 0, fmt.Errorf("failed to perform query: %w", err) + } + + if len(result) != 1 { + return 0, fmt.Errorf("unexpected number of rows returned, expected 1 got %d", len(result)) + } + + return result[0].Count, nil +} + +// Cleanup removes all connections that have ended before threshold. +// +// NOTE(ppacher): there is no easy way to get the number of removed +// rows other than counting them in a first step. Though, that's +// probably not worth the cylces... +func (db *Database) Cleanup(ctx context.Context, threshold time.Time) (int, error) { + where := `WHERE ended IS NOT NULL + AND datetime(ended) < datetime(:threshold)` + sql := "DELETE FROM connections " + where + ";" + + args := orm.WithNamedArgs(map[string]interface{}{ + ":threshold": threshold.UTC().Format(orm.SqliteTimeFormat), + }) + + var result []struct { + Count int `sqlite:"count"` + } + if err := db.Execute( + ctx, + "SELECT COUNT(*) AS count FROM connections "+where, + args, + orm.WithTransient(), + orm.WithResult(&result), + ); err != nil { + return 0, fmt.Errorf("failed to perform query: %w", err) + } + if len(result) != 1 { + return 0, fmt.Errorf("unexpected number of rows, expected 1 got %d", len(result)) + } + + err := db.Execute(ctx, sql, args) + if err != nil { + return 0, err + } + + return result[0].Count, nil +} + +// dumpTo is a simple helper method that dumps all rows stored in the SQLite database +// as JSON to w. +// Any error aborts dumping rows and is returned. +func (db *Database) dumpTo(ctx context.Context, w io.Writer) error { + db.l.Lock() + defer db.l.Unlock() + + var conns []Conn + if err := sqlitex.ExecuteTransient(db.conn, "SELECT * FROM connections", &sqlitex.ExecOptions{ + ResultFunc: func(stmt *sqlite.Stmt) error { + var c Conn + if err := orm.DecodeStmt(ctx, db.Schema, stmt, &c, orm.DefaultDecodeConfig); err != nil { + return err + } + + conns = append(conns, c) + return nil + }, + }); err != nil { + return err + } + + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + return enc.Encode(conns) +} + +// Save inserts the connection conn into the SQLite database. If conn +// already exists the table row is updated instead. +func (db *Database) Save(ctx context.Context, conn Conn) error { + connMap, err := orm.ToParamMap(ctx, conn, "", orm.DefaultEncodeConfig) + if err != nil { + return fmt.Errorf("failed to encode connection for SQL: %w", err) + } + + columns := make([]string, 0, len(connMap)) + placeholders := make([]string, 0, len(connMap)) + values := make(map[string]interface{}, len(connMap)) + updateSets := make([]string, 0, len(connMap)) + + for key, value := range connMap { + columns = append(columns, key) + placeholders = append(placeholders, ":"+key) + values[":"+key] = value + updateSets = append(updateSets, fmt.Sprintf("%s = :%s", key, key)) + } + + db.l.Lock() + defer db.l.Unlock() + + // TODO(ppacher): make sure this one can be cached to speed up inserting + // and save some CPU cycles for the user + sql := fmt.Sprintf( + `INSERT INTO connections (%s) + VALUES(%s) + ON CONFLICT(id) DO UPDATE SET + %s + `, + strings.Join(columns, ", "), + strings.Join(placeholders, ", "), + strings.Join(updateSets, ", "), + ) + + if err := sqlitex.ExecuteTransient(db.conn, sql, &sqlitex.ExecOptions{ + Named: values, + ResultFunc: func(stmt *sqlite.Stmt) error { + log.Errorf("netquery: got result statement with %d columns", stmt.ColumnCount()) + return nil + }, + }); err != nil { + log.Errorf("netquery: failed to execute:\n\t%q\n\treturned error was: %s\n\tparameters: %+v", sql, err, values) + return err + } + + return nil +} + +// Close closes the underlying database connection. db should and cannot be +// used after Close() has returned. +func (db *Database) Close() error { + return db.conn.Close() +} diff --git a/netquery/manager.go b/netquery/manager.go new file mode 100644 index 00000000..b6649483 --- /dev/null +++ b/netquery/manager.go @@ -0,0 +1,251 @@ +package netquery + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "time" + + "github.com/safing/portbase/database/record" + "github.com/safing/portbase/formats/dsd" + "github.com/safing/portbase/log" + "github.com/safing/portbase/runtime" + "github.com/safing/portmaster/network" +) + +type ( + // ConnectionStore describes the interface that is used by Manager + // to save new or updated connection objects. + // It is implemented by the *Database type of this package. + ConnectionStore interface { + // Save is called to perists the new or updated connection. If required, + // It's up the the implementation to figure out if the operation is an + // insert or an update. + // The ID of Conn is unique and can be trusted to never collide with other + // connections of the save device. + Save(context.Context, Conn) error + } + + // Manager handles new and updated network.Connections feeds and persists them + // at a connection store. + // Manager also registers itself as a runtime database and pushes updates to + // connections using the local format. + // Users should use this update feed rather than the deprecated "network:" database. + Manager struct { + store ConnectionStore + push runtime.PushFunc + runtimeReg *runtime.Registry + pushPrefix string + } +) + +// NewManager returns a new connection manager that persists all newly created or +// updated connections at store. +func NewManager(store ConnectionStore, pushPrefix string, reg *runtime.Registry) (*Manager, error) { + mng := &Manager{ + store: store, + runtimeReg: reg, + pushPrefix: pushPrefix, + } + + push, err := reg.Register(pushPrefix, runtime.SimpleValueGetterFunc(mng.runtimeGet)) + if err != nil { + return nil, err + } + mng.push = push + + return mng, nil +} + +func (mng *Manager) runtimeGet(keyOrPrefix string) ([]record.Record, error) { + // TODO(ppacher): + // we don't yet support querying using the runtime database here ... + // consider exposing connection from the database at least by ID. + // + // NOTE(ppacher): + // for debugging purposes use RuntimeQueryRunner to execute plain + // SQL queries against the database using portbase/database/runtime. + return nil, nil +} + +// HandleFeed starts reading new and updated connections from feed and persists them +// in the configured ConnectionStore. HandleFeed blocks until either ctx is cancelled +// or feed is closed. +// Any errors encountered when processing new or updated connections are logged but +// otherwise ignored. +// HandleFeed handles and persists updates one after each other! Depending on the system +// load the user might want to use a buffered channel for feed. +func (mng *Manager) HandleFeed(ctx context.Context, feed <-chan *network.Connection) { + for { + select { + case <-ctx.Done(): + return + + case conn, ok := <-feed: + if !ok { + return + } + + model, err := convertConnection(conn) + if err != nil { + log.Errorf("netquery: failed to convert connection %s to sqlite model: %s", conn.ID, err) + + continue + } + + log.Tracef("netquery: updating connection %s", conn.ID) + + if err := mng.store.Save(ctx, *model); err != nil { + log.Errorf("netquery: failed to save connection %s in sqlite database: %s", conn.ID, err) + + continue + } + + // we clone the record metadata from the connection + // into the new model so the portbase/database layer + // can handle NEW/UPDATE correctly. + cloned := conn.Meta().Duplicate() + + // push an update for the connection + if err := mng.pushConnUpdate(ctx, *cloned, *model); err != nil { + log.Errorf("netquery: failed to push update for conn %s via database system: %s", conn.ID, err) + } + } + } +} + +func (mng *Manager) pushConnUpdate(ctx context.Context, meta record.Meta, conn Conn) error { + blob, err := json.Marshal(conn) + if err != nil { + return fmt.Errorf("failed to marshal connection: %w", err) + } + + key := fmt.Sprintf("%s:%s%s", mng.runtimeReg.DatabaseName(), mng.pushPrefix, conn.ID) + wrapper, err := record.NewWrapper( + key, + &meta, + dsd.JSON, + blob, + ) + if err != nil { + return fmt.Errorf("failed to create record wrapper: %w", err) + } + + mng.push(wrapper) + return nil +} + +// convertConnection converts conn to the local representation used +// to persist the information in SQLite. convertConnection attempts +// to lock conn and may thus block for some time. +func convertConnection(conn *network.Connection) (*Conn, error) { + conn.Lock() + defer conn.Unlock() + direction := "outbound" + if conn.Inbound { + direction = "inbound" + } + + c := Conn{ + ID: genConnID(conn), + External: conn.External, + IPVersion: conn.IPVersion, + IPProtocol: conn.IPProtocol, + LocalIP: conn.LocalIP.String(), + LocalPort: conn.LocalPort, + Verdict: conn.Verdict, + Started: time.Unix(conn.Started, 0), + Tunneled: conn.Tunneled, + Encrypted: conn.Encrypted, + Internal: conn.Internal, + Direction: direction, + Type: ConnectionTypeToString[conn.Type], + ProfileID: conn.ProcessContext.Source + "/" + conn.ProcessContext.Profile, + Path: conn.ProcessContext.BinaryPath, + ProfileRevision: int(conn.ProfileRevisionCounter), + ProfileName: conn.ProcessContext.ProfileName, + } + + switch conn.Type { + case network.DNSRequest: + c.Type = "dns" + case network.IPConnection: + c.Type = "ip" + } + + switch conn.Verdict { + case network.VerdictAccept, network.VerdictRerouteToNameserver, network.VerdictRerouteToTunnel: + accepted := true + c.Allowed = &accepted + case network.VerdictUndecided, network.VerdictUndeterminable: + c.Allowed = nil + default: + allowed := false + c.Allowed = &allowed + } + + if conn.Ended > 0 { + ended := time.Unix(conn.Ended, 0) + c.Ended = &ended + c.Active = false + } else { + c.Active = true + } + + extraData := map[string]interface{}{ + "pid": conn.ProcessContext.PID, + } + + if conn.TunnelContext != nil { + extraData["tunnel"] = conn.TunnelContext + exitNode := conn.TunnelContext.GetExitNodeID() + c.ExitNode = &exitNode + c.SPNUsed = true + } + + if conn.DNSContext != nil { + extraData["dns"] = conn.DNSContext + } + + // TODO(ppacher): enable when TLS inspection is merged + // if conn.TLSContext != nil { + // extraData["tls"] = conn.TLSContext + // } + + if conn.Entity != nil { + extraData["cname"] = conn.Entity.CNAME + extraData["blockedByLists"] = conn.Entity.BlockedByLists + extraData["blockedEntities"] = conn.Entity.BlockedEntities + extraData["reason"] = conn.Reason + + c.RemoteIP = conn.Entity.IP.String() + c.RemotePort = conn.Entity.Port + c.Domain = conn.Entity.Domain + c.Country = conn.Entity.Country + c.ASN = conn.Entity.ASN + c.ASOwner = conn.Entity.ASOrg + c.Scope = conn.Entity.IPScope + if conn.Entity.Coordinates != nil { + c.Latitude = conn.Entity.Coordinates.Latitude + c.Longitude = conn.Entity.Coordinates.Longitude + } + } + + // pre-compute the JSON blob for the extra data column + // and assign it. + extraDataBlob, err := json.Marshal(extraData) + if err != nil { + return nil, fmt.Errorf("failed to marshal extra data: %w", err) + } + c.ExtraData = extraDataBlob + + return &c, nil +} + +func genConnID(conn *network.Connection) string { + data := conn.ID + "-" + time.Unix(conn.Started, 0).String() + hash := sha256.Sum256([]byte(data)) + return hex.EncodeToString(hash[:]) +} diff --git a/netquery/module_api.go b/netquery/module_api.go new file mode 100644 index 00000000..e5127b90 --- /dev/null +++ b/netquery/module_api.go @@ -0,0 +1,165 @@ +package netquery + +import ( + "context" + "fmt" + "time" + + "github.com/safing/portbase/api" + "github.com/safing/portbase/config" + "github.com/safing/portbase/database" + "github.com/safing/portbase/database/query" + "github.com/safing/portbase/log" + "github.com/safing/portbase/modules" + "github.com/safing/portbase/runtime" + "github.com/safing/portmaster/network" +) + +type Module struct { + *modules.Module + + db *database.Interface + sqlStore *Database + mng *Manager + feed chan *network.Connection +} + +func init() { + mod := new(Module) + mod.Module = modules.Register( + "netquery", + mod.Prepare, + mod.Start, + mod.Stop, + "api", + "network", + "database", + ) +} + +func (m *Module) Prepare() error { + var err error + + m.db = database.NewInterface(&database.Options{ + Local: true, + Internal: true, + }) + + m.sqlStore, err = NewInMemory() + if err != nil { + return fmt.Errorf("failed to create in-memory database: %w", err) + } + + m.mng, err = NewManager(m.sqlStore, "netquery/data/", runtime.DefaultRegistry) + if err != nil { + return fmt.Errorf("failed to create manager: %w", err) + } + + m.feed = make(chan *network.Connection, 1000) + + queryHander := &QueryHandler{ + Database: m.sqlStore, + IsDevMode: config.Concurrent.GetAsBool(config.CfgDevModeKey, false), + } + + chartHandler := &ChartHandler{ + Database: m.sqlStore, + } + + // FIXME(ppacher): use appropriate permissions for this + if err := api.RegisterEndpoint(api.Endpoint{ + Path: "netquery/query", + MimeType: "application/json", + Read: api.PermitUser, + Write: api.PermitUser, + BelongsTo: m.Module, + HandlerFunc: queryHander.ServeHTTP, + Name: "Query Connections", + Description: "Query the in-memory sqlite connection database.", + }); err != nil { + return fmt.Errorf("failed to register API endpoint: %w", err) + } + + if err := api.RegisterEndpoint(api.Endpoint{ + Path: "netquery/charts/connection-active", + MimeType: "application/json", + Read: api.PermitUser, + Write: api.PermitUser, + BelongsTo: m.Module, + HandlerFunc: chartHandler.ServeHTTP, + Name: "Active Connections Chart", + Description: "Query the in-memory sqlite connection database and return a chart of active connections.", + }); err != nil { + return fmt.Errorf("failed to register API endpoint: %w", err) + } + + return nil +} + +func (mod *Module) Start() error { + mod.StartServiceWorker("netquery-feeder", time.Second, func(ctx context.Context) error { + sub, err := mod.db.Subscribe(query.New("network:")) + if err != nil { + return fmt.Errorf("failed to subscribe to network tree: %w", err) + } + defer sub.Cancel() + + for { + select { + case <-ctx.Done(): + return nil + case rec, ok := <-sub.Feed: + if !ok { + return nil + } + + conn, ok := rec.(*network.Connection) + if !ok { + // This is fine as we also receive process updates on + // this channel. + continue + } + + mod.feed <- conn + } + } + }) + + mod.StartServiceWorker("netquery-persister", time.Second, func(ctx context.Context) error { + mod.mng.HandleFeed(ctx, mod.feed) + return nil + }) + + mod.StartServiceWorker("netquery-row-cleaner", time.Second, func(ctx context.Context) error { + for { + select { + case <-ctx.Done(): + return nil + case <-time.After(10 * time.Second): + threshold := time.Now().Add(-network.DeleteConnsAfterEndedThreshold) + count, err := mod.sqlStore.Cleanup(ctx, threshold) + if err != nil { + log.Errorf("netquery: failed to count number of rows in memory: %s", err) + } else { + log.Tracef("netquery: successfully removed %d old rows that ended before %s", count, threshold) + } + } + } + }) + + // for debugging, we provide a simple direct SQL query interface using + // the runtime database + // FIXME: Expose only in dev mode. + _, err := NewRuntimeQueryRunner(mod.sqlStore, "netquery/query/", runtime.DefaultRegistry) + if err != nil { + return fmt.Errorf("failed to set up runtime SQL query runner: %w", err) + } + + return nil +} + +func (mod *Module) Stop() error { + close(mod.feed) + + return nil +} diff --git a/netquery/orm/decoder.go b/netquery/orm/decoder.go new file mode 100644 index 00000000..6cc16f97 --- /dev/null +++ b/netquery/orm/decoder.go @@ -0,0 +1,482 @@ +package orm + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "log" + "reflect" + "strings" + "time" + + "zombiezen.com/go/sqlite" +) + +// Commonly used error messages when working with orm. +var ( + errStructExpected = errors.New("encode: can only encode structs to maps") + errStructPointerExpected = errors.New("decode: result must be pointer to a struct type or map[string]interface{}") + errUnexpectedColumnType = errors.New("decode: unexpected column type") +) + +// constants used when transforming data to and from sqlite. +var ( + // sqliteTimeFromat defines the string representation that is + // expected by SQLite DATETIME functions. + // Note that SQLite itself does not include support for a DATETIME + // column type. Instead, dates and times are stored either as INTEGER, + // TEXT or REAL. + // This package provides support for time.Time being stored as TEXT (using a + // preconfigured timezone; UTC by default) or as INTEGER (the user can choose between + // unixepoch and unixnano-epoch where the nano variant is not offically supported by + // SQLITE). + SqliteTimeFormat = "2006-01-02 15:04:05" +) + +type ( + + // Stmt describes the interface that must be implemented in order to + // be decodable to a struct type using DecodeStmt. This interface is implemented + // by *sqlite.Stmt. + Stmt interface { + ColumnCount() int + ColumnName(int) string + ColumnType(int) sqlite.ColumnType + ColumnText(int) string + ColumnBool(int) bool + ColumnFloat(int) float64 + ColumnInt(int) int + ColumnReader(int) *bytes.Reader + } + + // DecodeFunc is called for each non-basic type during decoding. + DecodeFunc func(colIdx int, colDef *ColumnDef, stmt Stmt, fieldDef reflect.StructField, outval reflect.Value) (interface{}, bool, error) + + DecodeConfig struct { + DecodeHooks []DecodeFunc + } +) + +// DecodeStmt decodes the current result row loaded in Stmt into the struct or map type result. +// Decoding hooks configured in cfg are executed before trying to decode basic types and may +// be specified to provide support for special types. +// See DatetimeDecoder() for an example of a DecodeHook that handles graceful time.Time conversion. +// +func DecodeStmt(ctx context.Context, schema *TableSchema, stmt Stmt, result interface{}, cfg DecodeConfig) error { + // make sure we got something to decode into ... + if result == nil { + return fmt.Errorf("%w, got %T", errStructPointerExpected, result) + } + + // fast path for decoding into a map + if mp, ok := result.(*map[string]interface{}); ok { + return decodeIntoMap(ctx, schema, stmt, mp, cfg) + } + + // make sure we got a pointer in result + if reflect.TypeOf(result).Kind() != reflect.Ptr { + return fmt.Errorf("%w, got %T", errStructPointerExpected, result) + } + + // make sure it's a poiter to a struct type + t := reflect.ValueOf(result).Elem().Type() + if t.Kind() != reflect.Struct { + return fmt.Errorf("%w, got %T", errStructPointerExpected, result) + } + + // if result is a nil pointer make sure to allocate some space + // for the resulting struct + resultValue := reflect.ValueOf(result) + if resultValue.IsNil() { + resultValue.Set( + reflect.New(t), + ) + } + + // we need access to the struct directly and not to the + // pointer. + target := reflect.Indirect(resultValue) + + // create a lookup map from field name (or sqlite:"" tag) + // to the field name + lm := make(map[string]string) + for i := 0; i < target.NumField(); i++ { + fieldType := t.Field(i) + + // skip unexported fields + if !fieldType.IsExported() { + continue + } + + lm[sqlColumnName(fieldType)] = fieldType.Name + } + + // iterate over all columns and assign them to the correct + // fields + for i := 0; i < stmt.ColumnCount(); i++ { + colName := stmt.ColumnName(i) + fieldName, ok := lm[colName] + if !ok { + // there's no target field for this column + // so we can skip it + continue + } + fieldType, _ := t.FieldByName(fieldName) + + value := target.FieldByName(fieldName) + + colType := stmt.ColumnType(i) + + // if the column is reported as NULL we keep + // the field as it is. + // TODO(ppacher): should we set it to nil here? + if colType == sqlite.TypeNull { + continue + } + + // if value is a nil pointer we need to allocate some memory + // first + if getKind(value) == reflect.Ptr && value.IsNil() { + storage := reflect.New(fieldType.Type.Elem()) + + value.Set(storage) + + // make sure value actually points the + // dereferenced target storage + value = storage.Elem() + } + + colDef := schema.GetColumnDef(colName) + + // execute all decode hooks but make sure we use decodeBasic() as the + // last one. + columnValue, err := runDecodeHooks( + i, + colDef, + stmt, + fieldType, + value, + append(cfg.DecodeHooks, decodeBasic()), + ) + if err != nil { + return err + } + + // if we don't have a converted value now we try to + // decode basic types + if columnValue == nil { + return fmt.Errorf("cannot decode column %d (type=%s)", i, colType) + } + + //log.Printf("valueTypeName: %s fieldName = %s value-orig = %s value = %s (%v) newValue = %s", value.Type().String(), fieldName, target.FieldByName(fieldName).Type(), value.Type(), value, columnValue) + + // convert it to the target type if conversion is possible + newValue := reflect.ValueOf(columnValue) + if newValue.Type().ConvertibleTo(value.Type()) { + newValue = newValue.Convert(value.Type()) + } + + // assign the new value to the struct field. + value.Set(newValue) + } + + return nil +} + +// DatetimeDecoder is capable of decoding sqlite INTEGER or TEXT storage classes into +// time.Time. For INTEGER storage classes, it supports 'unixnano' struct tag value to +// decide between Unix or UnixNano epoch timestamps. +// +// FIXME(ppacher): update comment about loc parameter and TEXT storage class parsing +// +func DatetimeDecoder(loc *time.Location) DecodeFunc { + return func(colIdx int, colDef *ColumnDef, stmt Stmt, fieldDef reflect.StructField, outval reflect.Value) (interface{}, bool, error) { + // if we have the column definition available we + // use the target go type from there. + outType := outval.Type() + + if colDef != nil { + outType = colDef.GoType + } + + // we only care about "time.Time" here + if outType.String() != "time.Time" || (colDef != nil && !colDef.IsTime) { + log.Printf("not decoding %s %v", outType, colDef) + return nil, false, nil + } + + switch stmt.ColumnType(colIdx) { + case sqlite.TypeInteger: + // stored as unix-epoch, if unixnano is set in the struct field tag + // we parse it with nano-second resolution + // TODO(ppacher): actually split the tag value at "," and search + // the slice for "unixnano" + if strings.Contains(fieldDef.Tag.Get("sqlite"), ",unixnano") { + return time.Unix(0, int64(stmt.ColumnInt(colIdx))), true, nil + } + + return time.Unix(int64(stmt.ColumnInt(colIdx)), 0), true, nil + + case sqlite.TypeText: + // stored ISO8601 but does not have any timezone information + // assigned so we always treat it as loc here. + t, err := time.ParseInLocation(SqliteTimeFormat, stmt.ColumnText(colIdx), loc) + if err != nil { + return nil, false, fmt.Errorf("failed to parse %q in %s: %w", stmt.ColumnText(colIdx), fieldDef.Name, err) + } + + return t, true, nil + + case sqlite.TypeFloat: + // stored as Julian day numbers + return nil, false, fmt.Errorf("REAL storage type not support for time.Time") + + case sqlite.TypeNull: + return nil, true, nil + + default: + return nil, false, fmt.Errorf("unsupported storage type for time.Time: %s", stmt.ColumnType(colIdx)) + } + } +} + +func decodeIntoMap(ctx context.Context, schema *TableSchema, stmt Stmt, mp *map[string]interface{}, cfg DecodeConfig) error { + if *mp == nil { + *mp = make(map[string]interface{}) + } + + for i := 0; i < stmt.ColumnCount(); i++ { + var x interface{} + + colDef := schema.GetColumnDef(stmt.ColumnName(i)) + + outVal := reflect.ValueOf(&x).Elem() + fieldType := reflect.StructField{} + if colDef != nil { + outVal = reflect.New(colDef.GoType).Elem() + fieldType = reflect.StructField{ + Type: colDef.GoType, + } + } + + val, err := runDecodeHooks( + i, + colDef, + stmt, + fieldType, + outVal, + append(cfg.DecodeHooks, decodeBasic()), + ) + if err != nil { + return fmt.Errorf("failed to decode column %s: %w", stmt.ColumnName(i), err) + } + + (*mp)[stmt.ColumnName(i)] = val + } + + return nil +} + +func decodeBasic() DecodeFunc { + return func(colIdx int, colDef *ColumnDef, stmt Stmt, fieldDef reflect.StructField, outval reflect.Value) (result interface{}, handled bool, err error) { + valueKind := getKind(outval) + colType := stmt.ColumnType(colIdx) + colName := stmt.ColumnName(colIdx) + + errInvalidType := fmt.Errorf("%w %s for column %s with field type %s", errUnexpectedColumnType, colType.String(), colName, outval.Type()) + + // if we have the column definition available we + // use the target go type from there. + if colDef != nil { + valueKind = normalizeKind(colDef.GoType.Kind()) + + // if we have a column defintion we try to convert the value to + // the actual Go-type that was used in the model. + // this is useful, for example, to ensure a []byte{} is always decoded into json.RawMessage + // or that type aliases like (type myInt int) are decoded into myInt instead of int + defer func() { + if handled { + t := reflect.New(colDef.GoType).Elem() + + if result == nil || reflect.ValueOf(result).IsZero() { + return + } + + if reflect.ValueOf(result).Type().ConvertibleTo(colDef.GoType) { + result = reflect.ValueOf(result).Convert(colDef.GoType).Interface() + } + t.Set(reflect.ValueOf(result)) + + result = t.Interface() + } + }() + } + + log.Printf("decoding %s into kind %s", colName, valueKind) + + if colType == sqlite.TypeNull { + if colDef != nil && colDef.Nullable { + return nil, true, nil + } + + if colDef != nil && !colDef.Nullable { + return reflect.New(colDef.GoType).Elem().Interface(), true, nil + } + + if outval.Kind() == reflect.Ptr { + return nil, true, nil + } + } + + switch valueKind { + case reflect.String: + if colType != sqlite.TypeText { + return nil, false, errInvalidType + } + return stmt.ColumnText(colIdx), true, nil + + case reflect.Bool: + // sqlite does not have a BOOL type, it rather stores a 1/0 in a column + // with INTEGER affinity. + if colType != sqlite.TypeInteger { + return nil, false, errInvalidType + } + return stmt.ColumnBool(colIdx), true, nil + + case reflect.Float64: + if colType != sqlite.TypeFloat { + return nil, false, errInvalidType + } + return stmt.ColumnFloat(colIdx), true, nil + + case reflect.Int, reflect.Uint: // getKind() normalizes all ints to reflect.Int/Uint because sqlite doesn't really care ... + if colType != sqlite.TypeInteger { + return nil, false, errInvalidType + } + + return stmt.ColumnInt(colIdx), true, nil + + case reflect.Slice: + if outval.Type().Elem().Kind() != reflect.Uint8 { + return nil, false, fmt.Errorf("slices other than []byte for BLOB are not supported") + } + + if colType != sqlite.TypeBlob { + return nil, false, errInvalidType + } + + columnValue, err := io.ReadAll(stmt.ColumnReader(colIdx)) + if err != nil { + return nil, false, fmt.Errorf("failed to read blob for column %s: %w", fieldDef.Name, err) + } + + return columnValue, true, nil + + case reflect.Interface: + var ( + t reflect.Type + x interface{} + ) + switch colType { + case sqlite.TypeBlob: + t = reflect.TypeOf([]byte{}) + columnValue, err := io.ReadAll(stmt.ColumnReader(colIdx)) + if err != nil { + return nil, false, fmt.Errorf("failed to read blob for column %s: %w", fieldDef.Name, err) + } + x = columnValue + + case sqlite.TypeFloat: + t = reflect.TypeOf(float64(0)) + x = stmt.ColumnFloat(colIdx) + + case sqlite.TypeInteger: + t = reflect.TypeOf(int(0)) + x = stmt.ColumnInt(colIdx) + + case sqlite.TypeText: + t = reflect.TypeOf(string("")) + x = stmt.ColumnText(colIdx) + + case sqlite.TypeNull: + t = nil + x = nil + + default: + return nil, false, fmt.Errorf("unsupported column type %s", colType) + } + + if t == nil { + return nil, true, nil + } + + target := reflect.New(t).Elem() + target.Set(reflect.ValueOf(x)) + + return target.Interface(), true, nil + + default: + return nil, false, fmt.Errorf("cannot decode into %s", valueKind) + } + } +} + +func sqlColumnName(fieldType reflect.StructField) string { + tagValue, hasTag := fieldType.Tag.Lookup("sqlite") + if !hasTag { + return fieldType.Name + } + + parts := strings.Split(tagValue, ",") + if parts[0] != "" { + return parts[0] + } + + return fieldType.Name +} + +// runDecodeHooks tries to decode the column value of stmt at index colIdx into outval by running all decode hooks. +// The first hook that returns a non-nil interface wins, other hooks will not be executed. If an error is +// returned by a decode hook runDecodeHooks stops the error is returned to the caller. +func runDecodeHooks(colIdx int, colDef *ColumnDef, stmt Stmt, fieldDef reflect.StructField, outval reflect.Value, hooks []DecodeFunc) (interface{}, error) { + for _, fn := range hooks { + res, end, err := fn(colIdx, colDef, stmt, fieldDef, outval) + if err != nil { + return res, err + } + + if end { + return res, nil + } + } + + return nil, nil +} + +// getKind returns the kind of value but normalized Int, Uint and Float varaints +// to their base type. +func getKind(val reflect.Value) reflect.Kind { + kind := val.Kind() + return normalizeKind(kind) +} + +func normalizeKind(kind reflect.Kind) reflect.Kind { + switch { + case kind >= reflect.Int && kind <= reflect.Int64: + return reflect.Int + case kind >= reflect.Uint && kind <= reflect.Uint64: + return reflect.Uint + case kind >= reflect.Float32 && kind <= reflect.Float64: + return reflect.Float64 + default: + return kind + } +} + +var DefaultDecodeConfig = DecodeConfig{ + DecodeHooks: []DecodeFunc{ + DatetimeDecoder(time.UTC), + }, +} diff --git a/netquery/orm/decoder_test.go b/netquery/orm/decoder_test.go new file mode 100644 index 00000000..5abd324e --- /dev/null +++ b/netquery/orm/decoder_test.go @@ -0,0 +1,573 @@ +package orm + +import ( + "bytes" + "context" + "encoding/json" + "log" + "reflect" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "zombiezen.com/go/sqlite" +) + +type testStmt struct { + columns []string + values []interface{} + types []sqlite.ColumnType +} + +func (ts testStmt) ColumnCount() int { return len(ts.columns) } +func (ts testStmt) ColumnName(i int) string { return ts.columns[i] } +func (ts testStmt) ColumnBool(i int) bool { return ts.values[i].(bool) } +func (ts testStmt) ColumnText(i int) string { return ts.values[i].(string) } +func (ts testStmt) ColumnFloat(i int) float64 { return ts.values[i].(float64) } +func (ts testStmt) ColumnInt(i int) int { return ts.values[i].(int) } +func (ts testStmt) ColumnReader(i int) *bytes.Reader { return bytes.NewReader(ts.values[i].([]byte)) } +func (ts testStmt) ColumnType(i int) sqlite.ColumnType { return ts.types[i] } + +// compile time check +var _ Stmt = new(testStmt) + +type exampleFieldTypes struct { + S string + I int + F float64 + B bool +} + +type examplePointerTypes struct { + S *string + I *int + F *float64 + B *bool +} + +type exampleStructTags struct { + S string `sqlite:"col_string"` + I int `sqlite:"col_int"` +} + +type exampleIntConv struct { + I8 int8 + I16 int16 + I32 int32 + I64 int64 + I int +} + +type exampleBlobTypes struct { + B []byte +} + +type exampleJSONRawTypes struct { + B json.RawMessage +} + +type exampleTimeTypes struct { + T time.Time + TP *time.Time +} + +type exampleInterface struct { + I interface{} + IP *interface{} +} + +func (ett *exampleTimeTypes) Equal(other interface{}) bool { + oett, ok := other.(*exampleTimeTypes) + if !ok { + return false + } + return ett.T.Equal(oett.T) && (ett.TP != nil && oett.TP != nil && ett.TP.Equal(*oett.TP)) || (ett.TP == nil && oett.TP == nil) +} + +type myInt int + +type exampleTimeNano struct { + T time.Time `sqlite:",unixnano"` +} + +func (etn *exampleTimeNano) Equal(other interface{}) bool { + oetn, ok := other.(*exampleTimeNano) + if !ok { + return false + } + return etn.T.Equal(oetn.T) +} + +func Test_Decoder(t *testing.T) { + ctx := context.TODO() + + refTime := time.Date(2022, time.February, 15, 9, 51, 00, 00, time.UTC) + + cases := []struct { + Desc string + Stmt testStmt + ColumnDef []ColumnDef + Result interface{} + Expected interface{} + }{ + { + "Decoding into nil is not allowed", + testStmt{ + columns: nil, + values: nil, + types: nil, + }, + nil, + nil, + nil, + }, + { + "Decoding into basic types", + testStmt{ + columns: []string{"S", "I", "F", "B"}, + types: []sqlite.ColumnType{ + sqlite.TypeText, + sqlite.TypeInteger, + sqlite.TypeFloat, + sqlite.TypeInteger, + }, + values: []interface{}{ + "string value", + 1, + 1.2, + true, + }, + }, + nil, + &exampleFieldTypes{}, + &exampleFieldTypes{ + S: "string value", + I: 1, + F: 1.2, + B: true, + }, + }, + { + "Decoding into basic types with different order", + testStmt{ + columns: []string{"I", "S", "B", "F"}, + types: []sqlite.ColumnType{ + sqlite.TypeInteger, + sqlite.TypeText, + sqlite.TypeInteger, + sqlite.TypeFloat, + }, + values: []interface{}{ + 1, + "string value", + true, + 1.2, + }, + }, + nil, + &exampleFieldTypes{}, + &exampleFieldTypes{ + S: "string value", + I: 1, + F: 1.2, + B: true, + }, + }, + { + "Decoding into basic types with missing values", + testStmt{ + columns: []string{"F", "B"}, + types: []sqlite.ColumnType{ + sqlite.TypeFloat, + sqlite.TypeInteger, + }, + values: []interface{}{ + 1.2, + true, + }, + }, + nil, + &exampleFieldTypes{}, + &exampleFieldTypes{ + F: 1.2, + B: true, + }, + }, + { + "Decoding into pointer types", + testStmt{ + columns: []string{"S", "I", "F", "B"}, + types: []sqlite.ColumnType{ + sqlite.TypeText, + sqlite.TypeInteger, + sqlite.TypeFloat, + sqlite.TypeInteger, + }, + values: []interface{}{ + "string value", + 1, + 1.2, + true, + }, + }, + nil, + &examplePointerTypes{}, + func() interface{} { + s := "string value" + i := 1 + f := 1.2 + b := true + + return &examplePointerTypes{ + S: &s, + I: &i, + F: &f, + B: &b, + } + }, + }, + { + "Decoding into pointer types with missing values", + testStmt{ + columns: []string{"S", "B"}, + types: []sqlite.ColumnType{ + sqlite.TypeText, + sqlite.TypeInteger, + sqlite.TypeFloat, + sqlite.TypeInteger, + }, + values: []interface{}{ + "string value", + true, + }, + }, + nil, + &examplePointerTypes{}, + func() interface{} { + s := "string value" + b := true + + return &examplePointerTypes{ + S: &s, + B: &b, + } + }, + }, + { + "Decoding into fields with struct tags", + testStmt{ + columns: []string{"col_string", "col_int"}, + types: []sqlite.ColumnType{ + sqlite.TypeText, + sqlite.TypeInteger, + }, + values: []interface{}{ + "string value", + 1, + }, + }, + nil, + &exampleStructTags{}, + &exampleStructTags{ + S: "string value", + I: 1, + }, + }, + { + "Decoding into correct int type", + testStmt{ + columns: []string{"I8", "I16", "I32", "I64", "I"}, + types: []sqlite.ColumnType{ + sqlite.TypeInteger, + sqlite.TypeInteger, + sqlite.TypeInteger, + sqlite.TypeInteger, + sqlite.TypeInteger, + }, + values: []interface{}{ + 1, + 1, + 1, + 1, + 1, + }, + }, + nil, + &exampleIntConv{}, + &exampleIntConv{ + 1, 1, 1, 1, 1, + }, + }, + { + "Handling NULL values for basic types", + testStmt{ + columns: []string{"S", "I", "F"}, + types: []sqlite.ColumnType{ + sqlite.TypeNull, + sqlite.TypeNull, + sqlite.TypeFloat, + }, + values: []interface{}{ + // we use nil here but actually that does not matter + nil, + nil, + 1.0, + }, + }, + nil, + &exampleFieldTypes{}, + &exampleFieldTypes{ + F: 1.0, + }, + }, + { + "Handling NULL values for pointer types", + testStmt{ + columns: []string{"S", "I", "F"}, + types: []sqlite.ColumnType{ + sqlite.TypeNull, + sqlite.TypeNull, + sqlite.TypeFloat, + }, + values: []interface{}{ + // we use nil here but actually that does not matter + nil, + nil, + 1.0, + }, + }, + nil, + &examplePointerTypes{}, + func() interface{} { + f := 1.0 + + return &examplePointerTypes{F: &f} + }, + }, + { + "Handling blob types", + testStmt{ + columns: []string{"B"}, + types: []sqlite.ColumnType{ + sqlite.TypeBlob, + }, + values: []interface{}{ + ([]byte)("hello world"), + }, + }, + nil, + &exampleBlobTypes{}, + &exampleBlobTypes{ + B: ([]byte)("hello world"), + }, + }, + { + "Handling blob types as json.RawMessage", + testStmt{ + columns: []string{"B"}, + types: []sqlite.ColumnType{ + sqlite.TypeBlob, + }, + values: []interface{}{ + ([]byte)("hello world"), + }, + }, + nil, + &exampleJSONRawTypes{}, + &exampleJSONRawTypes{ + B: (json.RawMessage)("hello world"), + }, + }, + { + "Handling time.Time and pointers to it", + testStmt{ + columns: []string{"T", "TP"}, + types: []sqlite.ColumnType{ + sqlite.TypeInteger, + sqlite.TypeInteger, + }, + values: []interface{}{ + int(refTime.Unix()), + int(refTime.Unix()), + }, + }, + nil, + &exampleTimeTypes{}, + &exampleTimeTypes{ + T: refTime, + TP: &refTime, + }, + }, + { + "Handling time.Time in nano-second resolution (struct tags)", + testStmt{ + columns: []string{"T", "TP"}, + types: []sqlite.ColumnType{ + sqlite.TypeInteger, + sqlite.TypeInteger, + }, + values: []interface{}{ + int(refTime.UnixNano()), + int(refTime.UnixNano()), + }, + }, + nil, + &exampleTimeNano{}, + &exampleTimeNano{ + T: refTime, + }, + }, + { + "Decoding into interface", + testStmt{ + columns: []string{"I", "IP"}, + types: []sqlite.ColumnType{ + sqlite.TypeText, + sqlite.TypeText, + }, + values: []interface{}{ + "value1", + "value2", + }, + }, + nil, + &exampleInterface{}, + func() interface{} { + var x interface{} + x = "value2" + + return &exampleInterface{ + I: "value1", + IP: &x, + } + }, + }, + { + "Decoding into map[string]interface{}", + testStmt{ + columns: []string{"I", "F", "S", "B"}, + types: []sqlite.ColumnType{ + sqlite.TypeInteger, + sqlite.TypeFloat, + sqlite.TypeText, + sqlite.TypeBlob, + }, + values: []interface{}{ + 1, + 1.1, + "string value", + []byte("blob value"), + }, + }, + nil, + new(map[string]interface{}), + &map[string]interface{}{ + "I": 1, + "F": 1.1, + "S": "string value", + "B": []byte("blob value"), + }, + }, + { + "Decoding using type-hints", + testStmt{ + columns: []string{"B", "T"}, + types: []sqlite.ColumnType{ + sqlite.TypeInteger, + sqlite.TypeText, + }, + values: []interface{}{ + true, + refTime.Format(SqliteTimeFormat), + }, + }, + []ColumnDef{ + { + Name: "B", + Type: sqlite.TypeInteger, + GoType: reflect.TypeOf(true), + }, + { + Name: "T", + Type: sqlite.TypeText, + GoType: reflect.TypeOf(time.Time{}), + IsTime: true, + }, + }, + new(map[string]interface{}), + &map[string]interface{}{ + "B": true, + "T": refTime, + }, + }, + { + "Decoding into type aliases", + testStmt{ + columns: []string{"B"}, + types: []sqlite.ColumnType{ + sqlite.TypeBlob, + }, + values: []interface{}{ + []byte(`{"foo": "bar}`), + }, + }, + []ColumnDef{ + { + Name: "B", + Type: sqlite.TypeBlob, + GoType: reflect.TypeOf(json.RawMessage(`{"foo": "bar}`)), + }, + }, + new(map[string]interface{}), + &map[string]interface{}{ + "B": json.RawMessage(`{"foo": "bar}`), + }, + }, + { + "Decoding into type aliases #2", + testStmt{ + columns: []string{"I"}, + types: []sqlite.ColumnType{sqlite.TypeInteger}, + values: []interface{}{ + 10, + }, + }, + []ColumnDef{ + { + Name: "I", + Type: sqlite.TypeInteger, + GoType: reflect.TypeOf(myInt(0)), + }, + }, + new(map[string]interface{}), + &map[string]interface{}{ + "I": myInt(10), + }, + }, + } + + for idx := range cases { + c := cases[idx] + t.Run(c.Desc, func(t *testing.T) { + //t.Parallel() + + log.Println(c.Desc) + err := DecodeStmt(ctx, &TableSchema{Columns: c.ColumnDef}, c.Stmt, c.Result, DefaultDecodeConfig) + if fn, ok := c.Expected.(func() interface{}); ok { + c.Expected = fn() + } + + if c.Expected == nil { + assert.Error(t, err, c.Desc) + } else { + assert.NoError(t, err, c.Desc) + + if equaler, ok := c.Expected.(interface{ Equal(x interface{}) bool }); ok { + assert.True(t, equaler.Equal(c.Result)) + } else { + assert.Equal(t, c.Expected, c.Result) + } + } + }) + } +} diff --git a/netquery/orm/encoder.go b/netquery/orm/encoder.go new file mode 100644 index 00000000..0bcea756 --- /dev/null +++ b/netquery/orm/encoder.go @@ -0,0 +1,232 @@ +package orm + +import ( + "context" + "fmt" + "reflect" + "time" + + "zombiezen.com/go/sqlite" +) + +type ( + EncodeFunc func(col *ColumnDef, valType reflect.Type, val reflect.Value) (interface{}, bool, error) + + EncodeConfig struct { + EncodeHooks []EncodeFunc + } +) + +// ToParamMap returns a map that contains the sqlite compatible value of each struct field of +// r using the sqlite column name as a map key. It either uses the name of the +// exported struct field or the value of the "sqlite" tag. +func ToParamMap(ctx context.Context, r interface{}, keyPrefix string, cfg EncodeConfig) (map[string]interface{}, error) { + // make sure we work on a struct type + val := reflect.Indirect(reflect.ValueOf(r)) + if val.Kind() != reflect.Struct { + return nil, fmt.Errorf("%w, got %T", errStructExpected, r) + } + + res := make(map[string]interface{}, val.NumField()) + + for i := 0; i < val.NumField(); i++ { + fieldType := val.Type().Field(i) + field := val.Field(i) + + // skip unexported fields + if !fieldType.IsExported() { + continue + } + + colDef, err := getColumnDef(fieldType) + if err != nil { + return nil, fmt.Errorf("failed to get column definition for %s: %w", fieldType.Name, err) + } + + x, found, err := runEncodeHooks( + colDef, + fieldType.Type, + field, + append( + cfg.EncodeHooks, + encodeBasic(), + ), + ) + if err != nil { + return nil, fmt.Errorf("failed to run encode hooks: %w", err) + } + + if !found { + if reflect.Indirect(field).IsValid() { + x = reflect.Indirect(field).Interface() + } + } + + res[keyPrefix+sqlColumnName(fieldType)] = x + + } + + return res, nil +} + +func EncodeValue(ctx context.Context, colDef *ColumnDef, val interface{}, cfg EncodeConfig) (interface{}, error) { + fieldValue := reflect.ValueOf(val) + fieldType := reflect.TypeOf(val) + + x, found, err := runEncodeHooks( + colDef, + fieldType, + fieldValue, + append( + cfg.EncodeHooks, + encodeBasic(), + ), + ) + if err != nil { + return nil, fmt.Errorf("failed to run encode hooks: %w", err) + } + + if !found { + if reflect.Indirect(fieldValue).IsValid() { + x = reflect.Indirect(fieldValue).Interface() + } + } + + return x, nil +} + +func encodeBasic() EncodeFunc { + return func(col *ColumnDef, valType reflect.Type, val reflect.Value) (interface{}, bool, error) { + kind := valType.Kind() + if kind == reflect.Ptr { + valType = valType.Elem() + kind = valType.Kind() + + if val.IsNil() { + if !col.Nullable { + // we need to set the zero value here since the column + // is not marked as nullable + return reflect.New(valType).Elem().Interface(), true, nil + } + + return nil, true, nil + } + + val = val.Elem() + } + + switch normalizeKind(kind) { + case reflect.String, + reflect.Float64, + reflect.Bool, + reflect.Int, + reflect.Uint: + // sqlite package handles conversion of those types + // already + return val.Interface(), true, nil + + case reflect.Slice: + if valType.Elem().Kind() == reflect.Uint8 { + // this is []byte + return val.Interface(), true, nil + } + fallthrough + + default: + return nil, false, fmt.Errorf("cannot convert value of kind %s for use in SQLite", kind) + } + } +} + +func DatetimeEncoder(loc *time.Location) EncodeFunc { + return func(colDef *ColumnDef, valType reflect.Type, val reflect.Value) (interface{}, bool, error) { + // if fieldType holds a pointer we need to dereference the value + ft := valType.String() + if valType.Kind() == reflect.Ptr { + ft = valType.Elem().String() + val = reflect.Indirect(val) + } + + // we only care about "time.Time" here + var t time.Time + if ft == "time.Time" { + // handle the zero time as a NULL. + if !val.IsValid() || val.IsZero() { + return nil, true, nil + } + + var ok bool + valInterface := val.Interface() + t, ok = valInterface.(time.Time) + if !ok { + return nil, false, fmt.Errorf("cannot convert reflect value to time.Time") + } + + } else if valType.Kind() == reflect.String && colDef.IsTime { + var err error + t, err = time.Parse(time.RFC3339, val.String()) + if err != nil { + return nil, false, fmt.Errorf("failed to parse time as RFC3339: %w", err) + } + + } else { + // we don't care ... + return nil, false, nil + } + + switch colDef.Type { + case sqlite.TypeInteger: + if colDef.UnixNano { + return t.UnixNano(), true, nil + } + return t.Unix(), true, nil + + case sqlite.TypeText: + str := t.In(loc).Format(SqliteTimeFormat) + + return str, true, nil + } + + return nil, false, fmt.Errorf("cannot store time.Time in %s", colDef.Type) + } +} + +func runEncodeHooks(colDef *ColumnDef, valType reflect.Type, val reflect.Value, hooks []EncodeFunc) (interface{}, bool, error) { + if valType == nil { + if !colDef.Nullable { + switch colDef.Type { + case sqlite.TypeBlob: + return []byte{}, true, nil + case sqlite.TypeFloat: + return 0.0, true, nil + case sqlite.TypeText: + return "", true, nil + case sqlite.TypeInteger: + return 0, true, nil + default: + return nil, false, fmt.Errorf("unsupported sqlite data type: %s", colDef.Type) + } + } + + return nil, true, nil + } + + for _, fn := range hooks { + res, end, err := fn(colDef, valType, val) + if err != nil { + return res, false, err + } + + if end { + return res, true, nil + } + } + + return nil, false, nil +} + +var DefaultEncodeConfig = EncodeConfig{ + EncodeHooks: []EncodeFunc{ + DatetimeEncoder(time.UTC), + }, +} diff --git a/netquery/orm/encoder_test.go b/netquery/orm/encoder_test.go new file mode 100644 index 00000000..aff28580 --- /dev/null +++ b/netquery/orm/encoder_test.go @@ -0,0 +1,260 @@ +package orm + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "zombiezen.com/go/sqlite" +) + +func Test_EncodeAsMap(t *testing.T) { + ctx := context.TODO() + refTime := time.Date(2022, time.February, 15, 9, 51, 00, 00, time.UTC) + + cases := []struct { + Desc string + Input interface{} + Expected map[string]interface{} + }{ + { + "Encode basic types", + struct { + I int + F float64 + S string + B []byte + }{ + I: 1, + F: 1.2, + S: "string", + B: ([]byte)("bytes"), + }, + map[string]interface{}{ + "I": 1, + "F": 1.2, + "S": "string", + "B": ([]byte)("bytes"), + }, + }, + { + "Encode using struct tags", + struct { + I int `sqlite:"col_int"` + S string `sqlite:"col_string"` + }{ + I: 1, + S: "string value", + }, + map[string]interface{}{ + "col_int": 1, + "col_string": "string value", + }, + }, + { + "Ignore Private fields", + struct { + I int + s string + }{ + I: 1, + s: "string value", + }, + map[string]interface{}{ + "I": 1, + }, + }, + { + "Handle Pointers", + struct { + I *int + S *string + }{ + I: new(int), + }, + map[string]interface{}{ + "I": 0, + "S": nil, + }, + }, + { + "Handle time.Time types", + struct { + TinInt time.Time `sqlite:",integer,unixnano"` + TinString time.Time `sqlite:",text"` + }{ + TinInt: refTime, + TinString: refTime, + }, + map[string]interface{}{ + "TinInt": refTime.UnixNano(), + "TinString": refTime.Format(SqliteTimeFormat), + }, + }, + { + "Handle time.Time pointer types", + struct { + TinInt *time.Time `sqlite:",integer,unixnano"` + TinString *time.Time `sqlite:",text"` + Tnil1 *time.Time `sqlite:",text"` + Tnil2 *time.Time `sqlite:",text"` + }{ + TinInt: &refTime, + TinString: &refTime, + Tnil1: nil, + Tnil2: (*time.Time)(nil), + }, + map[string]interface{}{ + "TinInt": refTime.UnixNano(), + "TinString": refTime.Format(SqliteTimeFormat), + "Tnil1": nil, + "Tnil2": nil, + }, + }, + } + + for idx := range cases { + c := cases[idx] + t.Run(c.Desc, func(t *testing.T) { + // t.Parallel() + + res, err := ToParamMap(ctx, c.Input, "", DefaultEncodeConfig) + assert.NoError(t, err) + assert.Equal(t, c.Expected, res) + }) + } +} + +func Test_EncodeValue(t *testing.T) { + ctx := context.TODO() + refTime := time.Date(2022, time.February, 15, 9, 51, 00, 00, time.UTC) + + cases := []struct { + Desc string + Column ColumnDef + Input interface{} + Output interface{} + }{ + { + "Special value time.Time as text", + ColumnDef{ + IsTime: true, + Type: sqlite.TypeText, + }, + refTime, + refTime.Format(SqliteTimeFormat), + }, + { + "Special value time.Time as unix-epoch", + ColumnDef{ + IsTime: true, + Type: sqlite.TypeInteger, + }, + refTime, + refTime.Unix(), + }, + { + "Special value time.Time as unixnano-epoch", + ColumnDef{ + IsTime: true, + Type: sqlite.TypeInteger, + UnixNano: true, + }, + refTime, + refTime.UnixNano(), + }, + { + "Special value zero time", + ColumnDef{ + IsTime: true, + Type: sqlite.TypeText, + }, + time.Time{}, + nil, + }, + { + "Special value zero time pointer", + ColumnDef{ + IsTime: true, + Type: sqlite.TypeText, + }, + new(time.Time), + nil, + }, + { + "Special value *time.Time as text", + ColumnDef{ + IsTime: true, + Type: sqlite.TypeText, + }, + &refTime, + refTime.Format(SqliteTimeFormat), + }, + { + "Special value untyped nil", + ColumnDef{ + Nullable: true, + IsTime: true, + Type: sqlite.TypeText, + }, + nil, + nil, + }, + { + "Special value typed nil", + ColumnDef{ + IsTime: true, + Type: sqlite.TypeText, + }, + (*time.Time)(nil), + nil, + }, + { + "Time formated as string", + ColumnDef{ + IsTime: true, + Type: sqlite.TypeText, + }, + refTime.In(time.Local).Format(time.RFC3339), + refTime.Format(SqliteTimeFormat), + }, + { + "Nullable integer", + ColumnDef{ + Type: sqlite.TypeInteger, + Nullable: true, + }, + nil, + nil, + }, + { + "Not-Null integer", + ColumnDef{ + Name: "test", + Type: sqlite.TypeInteger, + }, + nil, + 0, + }, + { + "Not-Null string", + ColumnDef{ + Type: sqlite.TypeText, + }, + nil, + "", + }, + } + + for idx := range cases { + c := cases[idx] + t.Run(c.Desc, func(t *testing.T) { + //t.Parallel() + + res, err := EncodeValue(ctx, &c.Column, c.Input, DefaultEncodeConfig) + assert.NoError(t, err) + assert.Equal(t, c.Output, res) + }) + } +} diff --git a/netquery/orm/query_runner.go b/netquery/orm/query_runner.go new file mode 100644 index 00000000..88eceefd --- /dev/null +++ b/netquery/orm/query_runner.go @@ -0,0 +1,171 @@ +package orm + +import ( + "context" + "fmt" + "reflect" + + "zombiezen.com/go/sqlite" + "zombiezen.com/go/sqlite/sqlitex" +) + +type ( + // QueryOption can be specified at RunQuery to alter the behavior + // of the executed query. + QueryOption func(opts *queryOpts) + + queryOpts struct { + Transient bool + Args []interface{} + NamedArgs map[string]interface{} + Result interface{} + DecodeConfig DecodeConfig + Schema TableSchema + } +) + +// WithTransient marks the query as transient. +// +// Transient queries will not be cached for later +// re-use after they have been prepared. +func WithTransient() QueryOption { + return func(opts *queryOpts) { + opts.Transient = true + } +} + +// WithArgs adds a list of arguments for the query. Arguments +// are applied in order. +// +// See SQL Language Expression documentation of SQLite for +// details: https://sqlite.org/lang_expr.html +func WithArgs(args ...interface{}) QueryOption { + return func(opts *queryOpts) { + opts.Args = args + } +} + +// WithNamedArgs adds args to the query. The query must used +// named argument placeholders. According to the SQLite spec, +// arguments must either start with ':', '@' or '$'. +// +// See SQL Language Expression documentation of SQLite for +// details: https://sqlite.org/lang_expr.html +func WithNamedArgs(args map[string]interface{}) QueryOption { + return func(opts *queryOpts) { + opts.NamedArgs = args + } +} + +func WithSchema(tbl TableSchema) QueryOption { + return func(opts *queryOpts) { + opts.Schema = tbl + } +} + +// WithResult sets the result receiver. result is expected to +// be a pointer to a slice of struct or map types. +// +// For decoding DecodeStmt is used to decode each +// row into a new slice element. It thus supports special values +// like time.Time. See DecodeStmt() and WithDecodeConfig() for +// more information. +func WithResult(result interface{}) QueryOption { + return func(opts *queryOpts) { + opts.Result = result + } +} + +// WithDecodeConfig configures the DecodeConfig to use when +// calling DecodeStmt to decode each row into the result slice. +// +// If not specified, DefaultDecodeConfig will be used. +func WithDecodeConfig(cfg DecodeConfig) QueryOption { + return func(opts *queryOpts) { + opts.DecodeConfig = cfg + } +} + +// RunQuery executes the query stored in sql against the databased opened in +// conn. Please refer to the documentation of QueryOption, especially WithResult() +// for more information on how to retrieve the resulting rows. +// +// Example: +// +// var result []struct{ +// Count int `sqlite:"rowCount"` +// } +// +// err := RunQuery(ctx, conn, "SELECT COUNT(*) AS rowCount FROM table", WithResult(&result)) +// fmt.Println(result[0].Count) +// +func RunQuery(ctx context.Context, conn *sqlite.Conn, sql string, modifiers ...QueryOption) error { + args := queryOpts{ + DecodeConfig: DefaultDecodeConfig, + } + + for _, fn := range modifiers { + fn(&args) + } + + opts := &sqlitex.ExecOptions{ + Args: args.Args, + Named: args.NamedArgs, + } + + var ( + sliceVal reflect.Value + valElemType reflect.Type + ) + + if args.Result != nil { + target := args.Result + outVal := reflect.ValueOf(target) + if outVal.Kind() != reflect.Ptr { + return fmt.Errorf("target must be a pointer, got %T", target) + } + + sliceVal = reflect.Indirect(outVal) + if !sliceVal.IsValid() || sliceVal.IsNil() { + newVal := reflect.Zero(outVal.Type().Elem()) + sliceVal.Set(newVal) + } + + kind := sliceVal.Kind() + if kind != reflect.Slice { + return fmt.Errorf("target but be pointer to slice, got %T", target) + } + valType := sliceVal.Type() + valElemType = valType.Elem() + + opts.ResultFunc = func(stmt *sqlite.Stmt) error { + var currentField reflect.Value + + currentField = reflect.New(valElemType) + + if err := DecodeStmt(ctx, &args.Schema, stmt, currentField.Interface(), args.DecodeConfig); err != nil { + return err + } + + sliceVal = reflect.Append(sliceVal, reflect.Indirect(currentField)) + + return nil + } + } + + var err error + if args.Transient { + err = sqlitex.ExecuteTransient(conn, sql, opts) + } else { + err = sqlitex.Execute(conn, sql, opts) + } + if err != nil { + return err + } + + if args.Result != nil { + reflect.Indirect(reflect.ValueOf(args.Result)).Set(sliceVal) + } + + return nil +} diff --git a/netquery/orm/schema_builder.go b/netquery/orm/schema_builder.go new file mode 100644 index 00000000..5d533be5 --- /dev/null +++ b/netquery/orm/schema_builder.go @@ -0,0 +1,235 @@ +package orm + +import ( + "errors" + "fmt" + "reflect" + "strconv" + "strings" + + "zombiezen.com/go/sqlite" +) + +var ( + errSkipStructField = errors.New("struct field should be skipped") +) + +var ( + TagUnixNano = "unixnano" + TagPrimaryKey = "primary" + TagAutoIncrement = "autoincrement" + TagTime = "time" + TagNotNull = "not-null" + TagNullable = "nullable" + TagTypeInt = "integer" + TagTypeText = "text" + TagTypePrefixVarchar = "varchar" + TagTypeBlob = "blob" + TagTypeFloat = "float" +) + +var sqlTypeMap = map[sqlite.ColumnType]string{ + sqlite.TypeBlob: "BLOB", + sqlite.TypeFloat: "REAL", + sqlite.TypeInteger: "INTEGER", + sqlite.TypeText: "TEXT", +} + +type ( + TableSchema struct { + Name string + Columns []ColumnDef + } + + ColumnDef struct { + Name string + Nullable bool + Type sqlite.ColumnType + GoType reflect.Type + Length int + PrimaryKey bool + AutoIncrement bool + UnixNano bool + IsTime bool + } +) + +func (ts TableSchema) GetColumnDef(name string) *ColumnDef { + for _, def := range ts.Columns { + if def.Name == name { + return &def + } + } + return nil +} + +func (ts TableSchema) CreateStatement(ifNotExists bool) string { + sql := "CREATE TABLE" + if ifNotExists { + sql += " IF NOT EXISTS" + } + sql += " " + ts.Name + " ( " + + for idx, col := range ts.Columns { + sql += col.AsSQL() + if idx < len(ts.Columns)-1 { + sql += ", " + } + } + + sql += " );" + return sql +} + +func (def ColumnDef) AsSQL() string { + sql := def.Name + " " + + if def.Type == sqlite.TypeText && def.Length > 0 { + sql += fmt.Sprintf("VARCHAR(%d)", def.Length) + } else { + sql += sqlTypeMap[def.Type] + } + + if def.PrimaryKey { + sql += " PRIMARY KEY" + } + if def.AutoIncrement { + sql += " AUTOINCREMENT" + } + if !def.Nullable { + sql += " NOT NULL" + } + + return sql +} + +func GenerateTableSchema(name string, d interface{}) (*TableSchema, error) { + ts := &TableSchema{ + Name: name, + } + + val := reflect.Indirect(reflect.ValueOf(d)) + if val.Kind() != reflect.Struct { + return nil, fmt.Errorf("%w, got %T", errStructExpected, d) + } + + for i := 0; i < val.NumField(); i++ { + fieldType := val.Type().Field(i) + if !fieldType.IsExported() { + continue + } + + def, err := getColumnDef(fieldType) + if err != nil { + if errors.Is(err, errSkipStructField) { + continue + } + + return nil, fmt.Errorf("struct field %s: %w", fieldType.Name, err) + } + + ts.Columns = append(ts.Columns, *def) + } + + return ts, nil +} + +func getColumnDef(fieldType reflect.StructField) (*ColumnDef, error) { + def := &ColumnDef{ + Name: fieldType.Name, + Nullable: fieldType.Type.Kind() == reflect.Ptr, + } + + ft := fieldType.Type + + if fieldType.Type.Kind() == reflect.Ptr { + ft = fieldType.Type.Elem() + } + + def.GoType = ft + kind := normalizeKind(ft.Kind()) + + switch kind { + case reflect.Int: + def.Type = sqlite.TypeInteger + + case reflect.Float64: + def.Type = sqlite.TypeFloat + + case reflect.String: + def.Type = sqlite.TypeText + + case reflect.Slice: + // only []byte/[]uint8 is supported + if ft.Elem().Kind() != reflect.Uint8 { + return nil, fmt.Errorf("slices of type %s is not supported", ft.Elem()) + } + + def.Type = sqlite.TypeBlob + } + + if err := applyStructFieldTag(fieldType, def); err != nil { + return nil, err + } + + return def, nil +} + +// applyStructFieldTag parses the sqlite:"" struct field tag and update the column +// definition def accordingly. +func applyStructFieldTag(fieldType reflect.StructField, def *ColumnDef) error { + parts := strings.Split(fieldType.Tag.Get("sqlite"), ",") + if len(parts) > 0 && parts[0] != "" { + if parts[0] == "-" { + return errSkipStructField + } + + def.Name = parts[0] + } + + if len(parts) > 1 { + for _, k := range parts[1:] { + switch k { + // column modifieres + case TagPrimaryKey: + def.PrimaryKey = true + case TagAutoIncrement: + def.AutoIncrement = true + case TagNotNull: + def.Nullable = false + case TagNullable: + def.Nullable = true + case TagUnixNano: + def.UnixNano = true + case TagTime: + def.IsTime = true + + // basic column types + case TagTypeInt: + def.Type = sqlite.TypeInteger + case TagTypeText: + def.Type = sqlite.TypeText + case TagTypeFloat: + def.Type = sqlite.TypeFloat + case TagTypeBlob: + def.Type = sqlite.TypeBlob + + // advanced column types + default: + if strings.HasPrefix(k, TagTypePrefixVarchar) { + lenStr := strings.TrimSuffix(strings.TrimPrefix(k, TagTypePrefixVarchar+"("), ")") + length, err := strconv.ParseInt(lenStr, 10, 0) + if err != nil { + return fmt.Errorf("failed to parse varchar length %q: %w", lenStr, err) + } + + def.Type = sqlite.TypeText + def.Length = int(length) + } + + } + } + } + + return nil +} diff --git a/netquery/orm/schema_builder_test.go b/netquery/orm/schema_builder_test.go new file mode 100644 index 00000000..7012076d --- /dev/null +++ b/netquery/orm/schema_builder_test.go @@ -0,0 +1,41 @@ +package orm + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func Test_SchemaBuilder(t *testing.T) { + cases := []struct { + Name string + Model interface{} + ExpectedSQL string + }{ + { + "Simple", + struct { + ID int `sqlite:"id,primary,autoincrement"` + Text string `sqlite:"text,nullable"` + Int *int `sqlite:",not-null"` + Float interface{} `sqlite:",float,nullable"` + }{}, + `CREATE TABLE Simple ( id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, text TEXT, Int INTEGER NOT NULL, Float REAL );`, + }, + { + "Varchar", + struct { + S string `sqlite:",varchar(10)"` + }{}, + `CREATE TABLE Varchar ( S VARCHAR(10) NOT NULL );`, + }, + } + + for idx := range cases { + c := cases[idx] + + res, err := GenerateTableSchema(c.Name, c.Model) + assert.NoError(t, err) + assert.Equal(t, c.ExpectedSQL, res.CreateStatement(false)) + } +} diff --git a/netquery/query.go b/netquery/query.go new file mode 100644 index 00000000..58c6cdee --- /dev/null +++ b/netquery/query.go @@ -0,0 +1,556 @@ +package netquery + +import ( + "context" + "encoding/json" + "fmt" + "io" + "log" + "sort" + "strings" + + "github.com/hashicorp/go-multierror" + "github.com/safing/portmaster/netquery/orm" + "zombiezen.com/go/sqlite" +) + +type ( + Query map[string][]Matcher + + MatchType interface { + Operator() string + } + + Equal interface{} + + Matcher struct { + Equal interface{} `json:"$eq,omitempty"` + NotEqual interface{} `json:"$ne,omitempty"` + In []interface{} `json:"$in,omitempty"` + NotIn []interface{} `json:"$notIn,omitempty"` + Like string `json:"$like,omitempty"` + } + + Count struct { + As string `json:"as"` + Field string `json:"field"` + Distinct bool `json:"distinct"` + } + + Sum struct { + Condition Query `json:"condition"` + As string `json:"as"` + Distinct bool `json:"distinct"` + } + + // NOTE: whenever adding support for new operators make sure + // to update UnmarshalJSON as well. + Select struct { + Field string `json:"field"` + Count *Count `json:"$count,omitempty"` + Sum *Sum `json:"$sum,omitempty"` + Distinct *string `json:"$distinct"` + } + + Selects []Select + + TextSearch struct { + Fields []string `json:"fields"` + Value string `json:"value"` + } + + QueryRequestPayload struct { + Select Selects `json:"select"` + Query Query `json:"query"` + OrderBy OrderBys `json:"orderBy"` + GroupBy []string `json:"groupBy"` + TextSearch *TextSearch `json:"textSearch"` + + Pagination + + selectedFields []string + whitelistedFields []string + paramMap map[string]interface{} + } + + QueryActiveConnectionChartPayload struct { + Query Query `json:"query"` + TextSearch *TextSearch `json:"textSearch"` + } + + OrderBy struct { + Field string `json:"field"` + Desc bool `json:"desc"` + } + + OrderBys []OrderBy + + Pagination struct { + PageSize int `json:"pageSize"` + Page int `json:"page"` + } +) + +func (query *Query) UnmarshalJSON(blob []byte) error { + if *query == nil { + *query = make(Query) + } + + var model map[string]json.RawMessage + + if err := json.Unmarshal(blob, &model); err != nil { + return err + } + + for columnName, rawColumnQuery := range model { + if len(rawColumnQuery) == 0 { + continue + } + + switch rawColumnQuery[0] { + case '{': + m, err := parseMatcher(rawColumnQuery) + if err != nil { + return err + } + + (*query)[columnName] = []Matcher{*m} + + case '[': + var rawMatchers []json.RawMessage + if err := json.Unmarshal(rawColumnQuery, &rawMatchers); err != nil { + return err + } + + (*query)[columnName] = make([]Matcher, len(rawMatchers)) + for idx, val := range rawMatchers { + // this should not happen + if len(val) == 0 { + continue + } + + // if val starts with a { we have a matcher definition + if val[0] == '{' { + m, err := parseMatcher(val) + if err != nil { + return err + } + (*query)[columnName][idx] = *m + + continue + } else if val[0] == '[' { + return fmt.Errorf("invalid token [ in query for column %s", columnName) + } + + // val is a dedicated JSON primitive and not an object or array + // so we treat that as an EQUAL condition. + var x interface{} + if err := json.Unmarshal(val, &x); err != nil { + return err + } + + (*query)[columnName][idx] = Matcher{ + Equal: x, + } + } + + default: + // value is a JSON primitive and not an object or array + // so we treat that as an EQUAL condition. + var x interface{} + if err := json.Unmarshal(rawColumnQuery, &x); err != nil { + return err + } + + (*query)[columnName] = []Matcher{ + {Equal: x}, + } + } + } + + return nil +} + +// TODO(ppacher): right now we only support LIMIT and OFFSET for pagination but that +// has an issue that loading the same page twice might yield different results due to +// new records shifting the result slice. To overcome this, return a "PageToken" to the +// user that includes the time the initial query was created so paginated queries can +// ensure new records don't end up in the result set. +func (page *Pagination) toSQLLimitOffsetClause() string { + limit := page.PageSize + + // default and cap the limit to at most 100 items + // per page to avoid out-of-memory conditions when loading + // thousands of results at once. + if limit <= 0 || limit > 100 { + limit = 100 + } + + sql := fmt.Sprintf("LIMIT %d", limit) + + if page.Page > 0 { + sql += fmt.Sprintf(" OFFSET %d", page.Page*limit) + } + + return sql +} + +func parseMatcher(raw json.RawMessage) (*Matcher, error) { + var m Matcher + if err := json.Unmarshal(raw, &m); err != nil { + return nil, err + } + + if err := m.Validate(); err != nil { + return nil, fmt.Errorf("invalid query matcher: %s", err) + } + log.Printf("parsed matcher %s: %+v", string(raw), m) + return &m, nil + +} + +func (match Matcher) Validate() error { + found := 0 + + if match.Equal != nil { + found++ + } + + if match.NotEqual != nil { + found++ + } + + if match.In != nil { + found++ + } + + if match.NotIn != nil { + found++ + } + + if match.Like != "" { + found++ + } + + if found == 0 { + return fmt.Errorf("no conditions specified") + } + + return nil +} + +func (text TextSearch) toSQLConditionClause(ctx context.Context, schema *orm.TableSchema, suffix string, encoderConfig orm.EncodeConfig) (string, map[string]interface{}, error) { + var ( + queryParts []string + params = make(map[string]interface{}) + ) + + key := fmt.Sprintf(":t%s", suffix) + params[key] = fmt.Sprintf("%%%s%%", text.Value) + + for _, field := range text.Fields { + colDef := schema.GetColumnDef(field) + if colDef == nil { + return "", nil, fmt.Errorf("column %s is not allowed in text-search", colDef.Name) + } + if colDef.Type != sqlite.TypeText { + return "", nil, fmt.Errorf("type of column %s cannot be used in text-search", colDef.Name) + } + + queryParts = append(queryParts, fmt.Sprintf("%s LIKE %s", colDef.Name, key)) + } + + if len(queryParts) == 0 { + return "", nil, nil + } + + return "( " + strings.Join(queryParts, " OR ") + " )", params, nil +} + +func (match Matcher) toSQLConditionClause(ctx context.Context, suffix string, conjunction string, colDef orm.ColumnDef, encoderConfig orm.EncodeConfig) (string, map[string]interface{}, error) { + var ( + queryParts []string + params = make(map[string]interface{}) + errs = new(multierror.Error) + key = fmt.Sprintf("%s%s", colDef.Name, suffix) + ) + + add := func(operator, suffix string, list bool, values ...interface{}) { + var placeholder []string + + for idx, value := range values { + encodedValue, err := orm.EncodeValue(ctx, &colDef, value, encoderConfig) + if err != nil { + errs.Errors = append(errs.Errors, + fmt.Errorf("failed to encode %v for column %s: %w", value, colDef.Name, err), + ) + return + } + + uniqKey := fmt.Sprintf(":%s%s%d", key, suffix, idx) + placeholder = append(placeholder, uniqKey) + params[uniqKey] = encodedValue + } + + if len(placeholder) == 1 && !list { + queryParts = append(queryParts, fmt.Sprintf("%s %s %s", colDef.Name, operator, placeholder[0])) + } else { + queryParts = append(queryParts, fmt.Sprintf("%s %s ( %s )", colDef.Name, operator, strings.Join(placeholder, ", "))) + } + } + + if match.Equal != nil { + add("=", "eq", false, match.Equal) + } + + if match.NotEqual != nil { + add("!=", "ne", false, match.NotEqual) + } + + if match.In != nil { + add("IN", "in", true, match.In...) + } + + if match.NotIn != nil { + add("NOT IN", "notin", true, match.NotIn...) + } + + if match.Like != "" { + add("LIKE", "like", false, match.Like) + } + + if len(queryParts) == 0 { + // this is an empty matcher without a single condition. + // we convert that to a no-op TRUE value + return "( 1 = 1 )", nil, errs.ErrorOrNil() + } + + if len(queryParts) == 1 { + return queryParts[0], params, errs.ErrorOrNil() + } + + return "( " + strings.Join(queryParts, " "+conjunction+" ") + " )", params, errs.ErrorOrNil() +} + +func (query Query) toSQLWhereClause(ctx context.Context, suffix string, m *orm.TableSchema, encoderConfig orm.EncodeConfig) (string, map[string]interface{}, error) { + if len(query) == 0 { + return "", nil, nil + } + + // create a lookup map to validate column names + lm := make(map[string]orm.ColumnDef, len(m.Columns)) + for _, col := range m.Columns { + lm[col.Name] = col + } + + paramMap := make(map[string]interface{}) + columnStmts := make([]string, 0, len(query)) + + // get all keys and sort them so we get a stable output + queryKeys := make([]string, 0, len(query)) + for column := range query { + queryKeys = append(queryKeys, column) + } + sort.Strings(queryKeys) + + // actually create the WHERE clause parts for each + // column in query. + errs := new(multierror.Error) + for _, column := range queryKeys { + values := query[column] + colDef, ok := lm[column] + if !ok { + errs.Errors = append(errs.Errors, fmt.Errorf("column %s is not allowed", column)) + + continue + } + + queryParts := make([]string, len(values)) + for idx, val := range values { + matcherQuery, params, err := val.toSQLConditionClause(ctx, fmt.Sprintf("%s%d", suffix, idx), "AND", colDef, encoderConfig) + if err != nil { + errs.Errors = append(errs.Errors, + fmt.Errorf("invalid matcher at index %d for column %s: %w", idx, colDef.Name, err), + ) + + continue + } + + // merge parameters up into the superior parameter map + for key, val := range params { + if _, ok := paramMap[key]; ok { + // is is soley a developer mistake when implementing a matcher so no forgiving ... + panic("sqlite parameter collision") + } + + paramMap[key] = val + } + + queryParts[idx] = matcherQuery + } + + columnStmts = append(columnStmts, + fmt.Sprintf("( %s )", strings.Join(queryParts, " OR ")), + ) + } + + whereClause := strings.Join(columnStmts, " AND ") + + return whereClause, paramMap, errs.ErrorOrNil() +} + +func (sel *Selects) UnmarshalJSON(blob []byte) error { + if len(blob) == 0 { + return io.ErrUnexpectedEOF + } + + // if we are looking at a slice directly decode into + // a []Select + if blob[0] == '[' { + var result []Select + if err := json.Unmarshal(blob, &result); err != nil { + return err + } + + (*sel) = result + + return nil + } + + // if it's an object decode into a single select + if blob[0] == '{' { + var result Select + if err := json.Unmarshal(blob, &result); err != nil { + return err + } + + *sel = []Select{result} + + return nil + } + + // otherwise this is just the field name + var field string + if err := json.Unmarshal(blob, &field); err != nil { + return err + } + + return nil +} + +func (sel *Select) UnmarshalJSON(blob []byte) error { + if len(blob) == 0 { + return io.ErrUnexpectedEOF + } + + // if we have an object at hand decode the select + // directly + if blob[0] == '{' { + var res struct { + Field string `json:"field"` + Count *Count `json:"$count"` + Sum *Sum `json:"$sum"` + Distinct *string `json:"$distinct"` + } + + if err := json.Unmarshal(blob, &res); err != nil { + return err + } + + sel.Count = res.Count + sel.Field = res.Field + sel.Distinct = res.Distinct + sel.Sum = res.Sum + + if sel.Count != nil && sel.Count.As != "" { + if !charOnlyRegexp.MatchString(sel.Count.As) { + return fmt.Errorf("invalid characters in $count.as, value must match [a-zA-Z]+") + } + } + + return nil + } + + var x string + if err := json.Unmarshal(blob, &x); err != nil { + return err + } + + sel.Field = x + + return nil +} + +func (orderBys *OrderBys) UnmarshalJSON(blob []byte) error { + if len(blob) == 0 { + return io.ErrUnexpectedEOF + } + + if blob[0] == '[' { + var result []OrderBy + if err := json.Unmarshal(blob, &result); err != nil { + return err + } + + *orderBys = result + + return nil + } + + if blob[0] == '{' { + var result OrderBy + if err := json.Unmarshal(blob, &result); err != nil { + return err + } + + *orderBys = []OrderBy{result} + + return nil + } + + var field string + if err := json.Unmarshal(blob, &field); err != nil { + return err + } + + *orderBys = []OrderBy{ + { + Field: field, + Desc: false, + }, + } + + return nil +} + +func (orderBy *OrderBy) UnmarshalJSON(blob []byte) error { + if len(blob) == 0 { + return io.ErrUnexpectedEOF + } + + if blob[0] == '{' { + var res struct { + Field string `json:"field"` + Desc bool `json:"desc"` + } + + if err := json.Unmarshal(blob, &res); err != nil { + return err + } + + orderBy.Desc = res.Desc + orderBy.Field = res.Field + + return nil + } + + var field string + if err := json.Unmarshal(blob, &field); err != nil { + return err + } + + orderBy.Field = field + orderBy.Desc = false + + return nil +} diff --git a/netquery/query_handler.go b/netquery/query_handler.go new file mode 100644 index 00000000..1b2b5411 --- /dev/null +++ b/netquery/query_handler.go @@ -0,0 +1,356 @@ +package netquery + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "io/ioutil" + "net/http" + "regexp" + "strings" + "time" + + "github.com/safing/portbase/log" + "github.com/safing/portmaster/netquery/orm" +) + +var ( + charOnlyRegexp = regexp.MustCompile("[a-zA-Z]+") +) + +type ( + + // QueryHandler implements http.Handler and allows to perform SQL + // query and aggregate functions on Database. + QueryHandler struct { + IsDevMode func() bool + Database *Database + } +) + +func (qh *QueryHandler) ServeHTTP(resp http.ResponseWriter, req *http.Request) { + start := time.Now() + requestPayload, err := qh.parseRequest(req) + if err != nil { + http.Error(resp, err.Error(), http.StatusBadRequest) + + return + } + + queryParsed := time.Since(start) + + query, paramMap, err := requestPayload.generateSQL(req.Context(), qh.Database.Schema) + if err != nil { + http.Error(resp, err.Error(), http.StatusBadRequest) + + return + } + + sqlQueryBuilt := time.Since(start) + + // actually execute the query against the database and collect the result + var result []map[string]interface{} + if err := qh.Database.Execute( + req.Context(), + query, + orm.WithNamedArgs(paramMap), + orm.WithResult(&result), + orm.WithSchema(*qh.Database.Schema), + ); err != nil { + http.Error(resp, "Failed to execute query: "+err.Error(), http.StatusInternalServerError) + + return + } + sqlQueryFinished := time.Since(start) + + // send the HTTP status code + resp.WriteHeader(http.StatusOK) + + // prepare the result encoder. + enc := json.NewEncoder(resp) + enc.SetEscapeHTML(false) + enc.SetIndent("", " ") + + // prepare the result body that, in dev mode, contains + // some diagnostics data about the query + var resultBody map[string]interface{} + if qh.IsDevMode() { + resultBody = map[string]interface{}{ + "sql_prep_stmt": query, + "sql_params": paramMap, + "query": requestPayload.Query, + "orderBy": requestPayload.OrderBy, + "groupBy": requestPayload.GroupBy, + "selects": requestPayload.Select, + "times": map[string]interface{}{ + "start_time": start, + "query_parsed_after": queryParsed.String(), + "query_built_after": sqlQueryBuilt.String(), + "query_executed_after": sqlQueryFinished.String(), + }, + } + } else { + resultBody = make(map[string]interface{}) + } + resultBody["results"] = result + + // and finally stream the response + if err := enc.Encode(resultBody); err != nil { + // we failed to encode the JSON body to resp so we likely either already sent a + // few bytes or the pipe was already closed. In either case, trying to send the + // error using http.Error() is non-sense. We just log it out here and that's all + // we can do. + log.Errorf("failed to encode JSON response: %s", err) + + return + } +} + +func (qh *QueryHandler) parseRequest(req *http.Request) (*QueryRequestPayload, error) { + var body io.Reader + + switch req.Method { + case http.MethodPost, http.MethodPut: + body = req.Body + case http.MethodGet: + body = strings.NewReader(req.URL.Query().Get("q")) + default: + return nil, fmt.Errorf("invalid HTTP method") + } + + var requestPayload QueryRequestPayload + blob, err := ioutil.ReadAll(body) + if err != nil { + return nil, fmt.Errorf("failed to read body" + err.Error()) + } + + body = bytes.NewReader(blob) + + dec := json.NewDecoder(body) + dec.DisallowUnknownFields() + + if err := json.Unmarshal(blob, &requestPayload); err != nil && !errors.Is(err, io.EOF) { + return nil, fmt.Errorf("invalid query: %w", err) + } + + return &requestPayload, nil +} + +func (req *QueryRequestPayload) generateSQL(ctx context.Context, schema *orm.TableSchema) (string, map[string]interface{}, error) { + if err := req.prepareSelectedFields(ctx, schema); err != nil { + return "", nil, fmt.Errorf("perparing selected fields: %w", err) + } + + // build the SQL where clause from the payload query + whereClause, paramMap, err := req.Query.toSQLWhereClause( + ctx, + "", + schema, + orm.DefaultEncodeConfig, + ) + if err != nil { + return "", nil, fmt.Errorf("generating where clause: %w", err) + } + + if req.paramMap == nil { + req.paramMap = make(map[string]interface{}) + } + + for key, val := range paramMap { + req.paramMap[key] = val + } + + if req.TextSearch != nil { + textClause, textParams, err := req.TextSearch.toSQLConditionClause(ctx, schema, "", orm.DefaultEncodeConfig) + if err != nil { + return "", nil, fmt.Errorf("generating text-search clause: %w", err) + } + + if textClause != "" { + if whereClause != "" { + whereClause += " AND " + } + + whereClause += textClause + + for key, val := range textParams { + req.paramMap[key] = val + } + } + } + + groupByClause, err := req.generateGroupByClause(schema) + if err != nil { + return "", nil, fmt.Errorf("generating group-by clause: %w", err) + } + + orderByClause, err := req.generateOrderByClause(schema) + if err != nil { + return "", nil, fmt.Errorf("generating order-by clause: %w", err) + } + + selectClause := req.generateSelectClause() + query := `SELECT ` + selectClause + ` FROM connections` + if whereClause != "" { + query += " WHERE " + whereClause + } + + query += " " + groupByClause + " " + orderByClause + " " + req.Pagination.toSQLLimitOffsetClause() + + return strings.TrimSpace(query), req.paramMap, nil +} + +func (req *QueryRequestPayload) prepareSelectedFields(ctx context.Context, schema *orm.TableSchema) error { + for idx, s := range req.Select { + var field string + switch { + case s.Count != nil: + field = s.Count.Field + case s.Distinct != nil: + field = *s.Distinct + case s.Sum != nil: + // field is not used in case of $sum + field = "*" + default: + field = s.Field + } + + colName := "*" + if field != "*" || (s.Count == nil && s.Sum == nil) { + var err error + + colName, err = req.validateColumnName(schema, field) + if err != nil { + return err + } + } + + switch { + case s.Count != nil: + var as = s.Count.As + if as == "" { + as = fmt.Sprintf("%s_count", colName) + } + var distinct = "" + if s.Count.Distinct { + distinct = "DISTINCT " + } + req.selectedFields = append( + req.selectedFields, + fmt.Sprintf("COUNT(%s%s) AS %s", distinct, colName, as), + ) + req.whitelistedFields = append(req.whitelistedFields, as) + + case s.Sum != nil: + if s.Sum.As == "" { + return fmt.Errorf("missing 'as' for $sum") + } + + clause, params, err := s.Sum.Condition.toSQLWhereClause(ctx, fmt.Sprintf("sel%d", idx), schema, orm.DefaultEncodeConfig) + if err != nil { + return fmt.Errorf("in $sum: %w", err) + } + + req.paramMap = params + req.selectedFields = append( + req.selectedFields, + fmt.Sprintf("SUM(%s) AS %s", clause, s.Sum.As), + ) + req.whitelistedFields = append(req.whitelistedFields, s.Sum.As) + + case s.Distinct != nil: + req.selectedFields = append(req.selectedFields, fmt.Sprintf("DISTINCT %s", colName)) + req.whitelistedFields = append(req.whitelistedFields, colName) + + default: + req.selectedFields = append(req.selectedFields, colName) + } + } + + return nil +} + +func (req *QueryRequestPayload) generateGroupByClause(schema *orm.TableSchema) (string, error) { + if len(req.GroupBy) == 0 { + return "", nil + } + + var groupBys = make([]string, len(req.GroupBy)) + + for idx, name := range req.GroupBy { + colName, err := req.validateColumnName(schema, name) + if err != nil { + return "", err + } + + groupBys[idx] = colName + } + + groupByClause := "GROUP BY " + strings.Join(groupBys, ", ") + + // if there are no explicitly selected fields we default to the + // group-by columns as that's what's expected most of the time anyway... + if len(req.selectedFields) == 0 { + req.selectedFields = append(req.selectedFields, groupBys...) + } + + return groupByClause, nil +} + +func (req *QueryRequestPayload) generateSelectClause() string { + var selectClause = "*" + if len(req.selectedFields) > 0 { + selectClause = strings.Join(req.selectedFields, ", ") + } + + return selectClause +} + +func (req *QueryRequestPayload) generateOrderByClause(schema *orm.TableSchema) (string, error) { + if len(req.OrderBy) == 0 { + return "", nil + } + + var orderBys = make([]string, len(req.OrderBy)) + for idx, sort := range req.OrderBy { + colName, err := req.validateColumnName(schema, sort.Field) + if err != nil { + return "", err + } + + if sort.Desc { + orderBys[idx] = fmt.Sprintf("%s DESC", colName) + } else { + orderBys[idx] = fmt.Sprintf("%s ASC", colName) + } + } + + return "ORDER BY " + strings.Join(orderBys, ", "), nil +} + +func (req *QueryRequestPayload) validateColumnName(schema *orm.TableSchema, field string) (string, error) { + colDef := schema.GetColumnDef(field) + if colDef != nil { + return colDef.Name, nil + } + + for _, selected := range req.whitelistedFields { + if field == selected { + return field, nil + } + } + + for _, selected := range req.selectedFields { + if field == selected { + return field, nil + } + } + + return "", fmt.Errorf("column name %q not allowed", field) +} + +// compile time check +var _ http.Handler = new(QueryHandler) diff --git a/netquery/query_test.go b/netquery/query_test.go new file mode 100644 index 00000000..7d9a0393 --- /dev/null +++ b/netquery/query_test.go @@ -0,0 +1,244 @@ +package netquery + +import ( + "context" + "encoding/json" + "fmt" + "testing" + "time" + + "github.com/safing/portmaster/netquery/orm" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func Test_UnmarshalQuery(t *testing.T) { + var cases = []struct { + Name string + Input string + Expected Query + Error error + }{ + { + "Parse a simple query", + `{ "domain": ["example.com", "example.at"] }`, + Query{ + "domain": []Matcher{ + { + Equal: "example.com", + }, + { + Equal: "example.at", + }, + }, + }, + nil, + }, + { + "Parse a more complex query", + ` + { + "domain": [ + { + "$in": [ + "example.at", + "example.com" + ] + }, + { + "$like": "microsoft.%" + } + ], + "path": [ + "/bin/ping", + { + "$notin": [ + "/sbin/ping", + "/usr/sbin/ping" + ] + } + ] + } + `, + Query{ + "domain": []Matcher{ + { + In: []interface{}{ + "example.at", + "example.com", + }, + }, + { + Like: "microsoft.%", + }, + }, + "path": []Matcher{ + { + Equal: "/bin/ping", + }, + { + NotIn: []interface{}{ + "/sbin/ping", + "/usr/sbin/ping", + }, + }, + }, + }, + nil, + }, + } + + for _, c := range cases { + t.Run(c.Name, func(t *testing.T) { + var q Query + err := json.Unmarshal([]byte(c.Input), &q) + + if c.Error != nil { + if assert.Error(t, err) { + assert.Equal(t, c.Error.Error(), err.Error()) + } + } else { + assert.NoError(t, err) + assert.Equal(t, c.Expected, q) + } + }) + } +} + +func Test_QueryBuilder(t *testing.T) { + now := time.Now() + + var cases = []struct { + N string + Q Query + R string + P map[string]interface{} + E error + }{ + { + "No filter", + nil, + "", + nil, + nil, + }, + { + "Simple, one-column filter", + Query{"domain": []Matcher{ + { + Equal: "example.com", + }, + { + Equal: "example.at", + }, + }}, + "( domain = :domain0eq0 OR domain = :domain1eq0 )", + map[string]interface{}{ + ":domain0eq0": "example.com", + ":domain1eq0": "example.at", + }, + nil, + }, + { + "Two column filter", + Query{ + "domain": []Matcher{ + { + Equal: "example.com", + }, + }, + "path": []Matcher{ + { + Equal: "/bin/curl", + }, + { + Equal: "/bin/ping", + }, + }, + }, + "( domain = :domain0eq0 ) AND ( path = :path0eq0 OR path = :path1eq0 )", + map[string]interface{}{ + ":domain0eq0": "example.com", + ":path0eq0": "/bin/curl", + ":path1eq0": "/bin/ping", + }, + nil, + }, + { + "Time based filter", + Query{ + "started": []Matcher{ + { + Equal: now.Format(time.RFC3339), + }, + }, + }, + "( started = :started0eq0 )", + map[string]interface{}{ + ":started0eq0": now.In(time.UTC).Format(orm.SqliteTimeFormat), + }, + nil, + }, + { + "Invalid column access", + Query{ + "forbiddenField": []Matcher{{}}, + }, + "", + nil, + fmt.Errorf("1 error occurred:\n\t* column forbiddenField is not allowed\n\n"), + }, + { + "Complex example", + Query{ + "domain": []Matcher{ + { + In: []interface{}{"example.at", "example.com"}, + }, + { + Like: "microsoft.%", + }, + }, + "path": []Matcher{ + { + NotIn: []interface{}{ + "/bin/ping", + "/sbin/ping", + "/usr/bin/ping", + }, + }, + }, + }, + "( domain IN ( :domain0in0, :domain0in1 ) OR domain LIKE :domain1like0 ) AND ( path NOT IN ( :path0notin0, :path0notin1, :path0notin2 ) )", + map[string]interface{}{ + ":domain0in0": "example.at", + ":domain0in1": "example.com", + ":domain1like0": "microsoft.%", + ":path0notin0": "/bin/ping", + ":path0notin1": "/sbin/ping", + ":path0notin2": "/usr/bin/ping", + }, + nil, + }, + } + + tbl, err := orm.GenerateTableSchema("connections", Conn{}) + require.NoError(t, err) + + for idx, c := range cases { + t.Run(c.N, func(t *testing.T) { + //t.Parallel() + str, params, err := c.Q.toSQLWhereClause(context.TODO(), "", tbl, orm.DefaultEncodeConfig) + + if c.E != nil { + if assert.Error(t, err) { + assert.Equal(t, c.E.Error(), err.Error(), "test case %d", idx) + } + } else { + assert.NoError(t, err, "test case %d", idx) + assert.Equal(t, c.P, params, "test case %d", idx) + assert.Equal(t, c.R, str, "test case %d", idx) + } + }) + } +} diff --git a/netquery/runtime_query_runner.go b/netquery/runtime_query_runner.go new file mode 100644 index 00000000..3b443ec5 --- /dev/null +++ b/netquery/runtime_query_runner.go @@ -0,0 +1,77 @@ +package netquery + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/safing/portbase/database/record" + "github.com/safing/portbase/formats/dsd" + "github.com/safing/portbase/log" + "github.com/safing/portbase/runtime" + "github.com/safing/portmaster/netquery/orm" +) + +// RuntimeQueryRunner provides a simple interface for the runtime database +// that allows direct SQL queries to be performed against db. +// Each resulting row of that query are marshaled as map[string]interface{} +// and returned as a single record to the caller. +// +// Using portbase/database#Query is not possible because portbase/database will +// complain about the SQL query being invalid. To work around that issue, +// RuntimeQueryRunner uses a 'GET key' request where the SQL query is embedded into +// the record key. +type RuntimeQueryRunner struct { + db *Database + reg *runtime.Registry + keyPrefix string +} + +// NewRuntimeQueryRunner returns a new runtime SQL query runner that parses +// and serves SQL queries form GET / requests. +func NewRuntimeQueryRunner(db *Database, prefix string, reg *runtime.Registry) (*RuntimeQueryRunner, error) { + runner := &RuntimeQueryRunner{ + db: db, + reg: reg, + keyPrefix: prefix, + } + + if _, err := reg.Register(prefix, runtime.SimpleValueGetterFunc(runner.get)); err != nil { + return nil, fmt.Errorf("failed to register runtime value provider: %w", err) + } + + return runner, nil +} + +func (runner *RuntimeQueryRunner) get(keyOrPrefix string) ([]record.Record, error) { + query := strings.TrimPrefix( + keyOrPrefix, + runner.keyPrefix, + ) + + log.Infof("netquery: executing custom SQL query: %q", query) + + var result []map[string]interface{} + if err := runner.db.Execute(context.Background(), query, orm.WithResult(&result)); err != nil { + return nil, fmt.Errorf("failed to perform query %q: %w", query, err) + } + + // we need to wrap the result slice into a map as portbase/database attempts + // to inject a _meta field. + blob, err := json.Marshal(map[string]interface{}{ + "result": result, + }) + if err != nil { + return nil, fmt.Errorf("failed to marshal result: %w", err) + } + + // construct a new record wrapper that uses the already prepared JSON blob. + key := fmt.Sprintf("%s:%s", runner.reg.DatabaseName(), keyOrPrefix) + wrapper, err := record.NewWrapper(key, new(record.Meta), dsd.JSON, blob) + if err != nil { + return nil, fmt.Errorf("failed to create record wrapper: %w", err) + } + + return []record.Record{wrapper}, nil +} diff --git a/network/clean.go b/network/clean.go index 608e2c88..6ff1eb38 100644 --- a/network/clean.go +++ b/network/clean.go @@ -12,7 +12,7 @@ import ( const ( cleanerTickDuration = 5 * time.Second - deleteConnsAfterEndedThreshold = 10 * time.Minute + DeleteConnsAfterEndedThreshold = 10 * time.Minute ) func connectionCleaner(ctx context.Context) error { @@ -41,7 +41,7 @@ func cleanConnections() (activePIDs map[int]struct{}) { _ = module.RunMediumPriorityMicroTask(&name, func(ctx context.Context) error { now := time.Now().UTC() nowUnix := now.Unix() - deleteOlderThan := now.Add(-deleteConnsAfterEndedThreshold).Unix() + deleteOlderThan := now.Add(-DeleteConnsAfterEndedThreshold).Unix() // network connections for _, conn := range conns.clone() { diff --git a/network/connection.go b/network/connection.go index 2756cffa..57ad93d7 100644 --- a/network/connection.go +++ b/network/connection.go @@ -149,7 +149,10 @@ type Connection struct { //nolint:maligned // TODO: fix alignment DNSContext *resolver.DNSRequestContext // TunnelContext holds additional information about the tunnel that this // connection is using. - TunnelContext interface{} + TunnelContext interface { + GetExitNodeID() string + } + // Internal is set to true if the connection is attributed as an // Portmaster internal connection. Internal may be set at different // points and access to it must be guarded by the connection lock.