yggdrasil-network/yggdrasil-go
1
0
Fork 0
mirror of https://github.com/yggdrasil-network/yggdrasil-go.git synced 2025-08-24 16:05:07 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'develop' into coords_info

Browse source
This commit is contained in:
vadym 2022-12-22 13:52:57 +02:00
commit 71b89bb67c
11 changed files with 422 additions and 293 deletions
Download patch file Download diff file Expand all files Collapse all files

10
.github/workflows/ci.yml vendored
View file

@ -18,11 +18,11 @@ jobs:
- uses: actions/setup-go@v3
with:
go-version: 1.19
- uses: actions/checkout@v3
- name: golangci-lint
uses: golangci/golangci-lint-action@v3
with:
args: --issues-exit-code=1
# - uses: actions/checkout@v3
# - name: golangci-lint
# uses: golangci/golangci-lint-action@v3
# with:
# args: --issues-exit-code=1
codeql:
name: Analyse

39
cmd/mesh/main.go
View file

@ -28,15 +28,17 @@ import (
"github.com/RiV-chain/RiV-mesh/src/core"
//"github.com/RiV-chain/RiV-mesh/src/ipv6rwc"
"github.com/RiV-chain/RiV-mesh/src/multicast"
"github.com/RiV-chain/RiV-mesh/src/restapi"
"github.com/RiV-chain/RiV-mesh/src/tun"
"github.com/RiV-chain/RiV-mesh/src/version"
)
type node struct {
core *core.Core
tun *tun.TunAdapter
multicast *multicast.Multicast
admin *admin.AdminSocket
core *core.Core
tun *tun.TunAdapter
multicast *multicast.Multicast
admin *admin.AdminSocket
rest_server *restapi.RestServer
}
func setLogLevel(loglevel string, logger *log.Logger) {
@ -127,6 +129,14 @@ func run(args yggArgs, ctx context.Context) {
default:
if logfd, err := os.OpenFile(args.logto, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644); err == nil {
logger = log.New(logfd, "", log.Flags())
defer func() int {
if r := recover(); r != nil {
logger.Println("Fatal error:", r)
fmt.Print(logfd)
return 1
}
return 0
}()
}
}
if logger == nil {
@ -262,6 +272,24 @@ func run(args yggArgs, ctx context.Context) {
}
}
// Setup the REST socket.
{
if n.rest_server, err = restapi.NewRestServer(restapi.RestServerCfg{
Core: n.core,
Log: logger,
ListenAddress: cfg.HttpAddress,
WwwRoot: cfg.WwwRoot,
ConfigFn: args.useconffile,
}); err != nil {
logger.Errorln(err)
} else {
err = n.rest_server.Serve()
if err != nil {
logger.Errorln(err)
}
}
}
// Setup the admin socket.
{
options := []admin.SetupOption{
@ -275,9 +303,6 @@ func run(args yggArgs, ctx context.Context) {
}
}
// Start HTTP server
n.admin.StartHttpServer(args.useconffile, cfg)
// Setup the multicast module.
{
options := []multicast.SetupOption{}

41
contrib/ui/mesh-ui/ui/assets/mesh-ui-es5.js
View file

@ -234,7 +234,7 @@ function togglePrivKeyVisibility() {
function humanReadableSpeed(speed) {
if (speed < 0) return "? B/s";
var i = speed < 1 ? 0 : Math.floor(Math.log(speed) / Math.log(1024));
return (speed / Math.pow(1024, i)).toFixed(2) * 1 + ' ' + ['bps', 'kbps', 'Mbps', 'Gbps', 'Tbps'][i];
return (speed / Math.pow(1024, i)).toFixed(2) * 1 + ' ' + ['B/s', 'kB/s', 'MB/s', 'GB/s', 'TB/s'][i];
}
var ui = ui || {
@ -297,22 +297,22 @@ ui.getConnectedPeers = function () {
});
};
var regexMulticast = /:\/\/\[fe80::/;
ui.updateConnectedPeersHandler = function (peers) {
ui.updateStatus(peers);
ui.updateSpeed(peers);
ui.updateCoordsInfo();
$("peers").innerText = "";
var regexStrip = /%[^\]]*/gm;
var sorted = peers.map(function (peer) {
return { "url": peer["remote"], "isMulticast": peer["remote"].match(regexMulticast) };
}).sort(function (a, b) {
return a.isMulticast > b.isMulticast;
});
sorted.forEach(function (peer) {
var row = $("peers").appendChild(document.createElement('div'));
row.className = "overflow-ellipsis";
var flag = row.appendChild(document.createElement("span"));
if (peer.isMulticast) flag.className = "fa fa-thin fa-share-nodes peer-connected-fl";else flag.className = "fi fi-" + ui.lookupCountryCodeByAddress(peer.url) + " peer-connected-fl";
row.append(peer.url.replace(regexStrip, ""));
});
if (peers) {
var regexStrip = /%[^\]]*/gm;
peers.forEach(function (peer) {
var row = $("peers").appendChild(document.createElement('div'));
row.className = "overflow-ellipsis";
var flag = row.appendChild(document.createElement("span"));
if (peer.multicast) flag.className = "fa fa-thin fa-share-nodes peer-connected-fl";
else flag.className = "fi fi-" + ui.lookupCountryCodeByAddress(peer.remote) + " peer-connected-fl";
row.append(peer.remote.replace(regexStrip, ""));
});
}
};
ui.updateStatus = function (peers) {
@ -320,7 +320,7 @@ ui.updateStatus = function (peers) {
if (peers) {
if (peers.length) {
var isNonMulticastExists = peers.filter(function (peer) {
return !peer["remote"].match(regexMulticast);
return !peer.multicast;
}).length;
status = isNonMulticastExists ? "st-multicast" : "st-connected";
} else {
@ -356,14 +356,10 @@ ui.updateSpeed = function (peers) {
ui.updateConnectedPeers = function () {
return ui.getConnectedPeers().then(function (peers) {
ui.updateConnectedPeersHandler(peers);
ui.updateStatus(peers);
ui.updateSpeed(peers);
return ui.updateConnectedPeersHandler(peers);
}).catch(function (error) {
ui.updateConnectedPeersHandler();
$("peers").innerText = error.message;
ui.updateStatus();
ui.updateSpeed();
ui.updateCoordsInfo();
});
};
@ -413,7 +409,6 @@ function main() {
ui.getAllPeers().then(function () {
return ui.updateConnectedPeers();
});
setInterval(ui.updateConnectedPeers, 5000);
ui.updateSelfInfo();

44
contrib/ui/mesh-ui/ui/assets/mesh-ui.js
View file

@ -227,7 +227,7 @@ function togglePrivKeyVisibility() {
function humanReadableSpeed(speed) {
if (speed < 0) return "? B/s";
var i = speed < 1 ? 0 : Math.floor(Math.log(speed) / Math.log(1024));
return (speed / Math.pow(1024, i)).toFixed(2) * 1 + ' ' + ['bps', 'kbps', 'Mbps', 'Gbps', 'Tbps'][i];
return (speed / Math.pow(1024, i)).toFixed(2) * 1 + ' ' + ['B/s', 'kB/s', 'MB/s', 'GB/s', 'TB/s'][i];
}
var ui = ui || {
@ -284,29 +284,31 @@ ui.getConnectedPeers = () =>
fetch('api/peers')
.then((response) => response.json())
const regexMulticast = /:\/\/\[fe80::/;
ui.updateConnectedPeersHandler = (peers) => {
ui.updateStatus(peers);
ui.updateSpeed(peers);
ui.updateCoordsInfo();
$("peers").innerText = "";
const regexStrip = /%[^\]]*/gm;
const sorted = peers.map(peer => ({"url": peer["remote"], "isMulticast": peer["remote"].match(regexMulticast)}))
.sort((a, b) => a.isMulticast > b.isMulticast);
sorted.forEach(peer => {
let row = $("peers").appendChild(document.createElement('div'));
row.className = "overflow-ellipsis"
let flag = row.appendChild(document.createElement("span"));
if(peer.isMulticast)
flag.className = "fa fa-thin fa-share-nodes peer-connected-fl";
else
flag.className = "fi fi-" + ui.lookupCountryCodeByAddress(peer.url) + " peer-connected-fl";
row.append(peer.url.replace(regexStrip, ""));
});
if(peers) {
const regexStrip = /%[^\]]*/gm;
peers.forEach(peer => {
let row = $("peers").appendChild(document.createElement('div'));
row.className = "overflow-ellipsis"
let flag = row.appendChild(document.createElement("span"));
if(peer.multicast)
flag.className = "fa fa-thin fa-share-nodes peer-connected-fl";
else
flag.className = "fi fi-" + ui.lookupCountryCodeByAddress(peer.remote) + " peer-connected-fl";
row.append(peer.remote.replace(regexStrip, ""));
});
}
}
ui.updateStatus = peers => {
let status = "st-error";
if(peers) {
if(peers.length) {
const isNonMulticastExists = peers.filter(peer => !peer["remote"].match(regexMulticast)).length;
const isNonMulticastExists = peers.filter(peer => !peer.multicast).length;
status = isNonMulticastExists ? "st-multicast" : "st-connected";
} else {
status = "st-connecting"
@ -335,15 +337,10 @@ ui.updateSpeed = peers => {
ui.updateConnectedPeers = () =>
ui.getConnectedPeers()
.then(peers => {ui.updateConnectedPeersHandler(peers);
ui.updateStatus(peers);
ui.updateSpeed(peers);
})
.then(peers => ui.updateConnectedPeersHandler(peers))
.catch((error) => {
ui.updateConnectedPeersHandler();
$("peers").innerText = error.message;
ui.updateStatus();
ui.updateSpeed();
ui.updateCoordsInfo();
});
ui.lookupCountryCodeByAddress = (address) => {
@ -387,7 +384,6 @@ function main() {
$("showAllPeersBtn").addEventListener("click", ui.showAllPeers);
ui.getAllPeers().then(() => ui.updateConnectedPeers());
setInterval(ui.updateConnectedPeers, 5000);
ui.updateSelfInfo();
//setInterval(ui.updateSelfInfo, 5000);

5
go.mod
View file

@ -28,7 +28,10 @@ require (
require github.com/webview/webview v0.0.0-20221220082822-77e021440a0f
require github.com/jchv/go-winloader v0.0.0-20210711035445-715c2860da7e // indirect
require (
github.com/jchv/go-winloader v0.0.0-20210711035445-715c2860da7e // indirect
github.com/vorot93/golang-signals v0.0.0-20170221070717-d9e83421ce45 // indirect
)
require (
github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815

2
go.sum
View file

@ -98,6 +98,8 @@ github.com/vishvananda/netlink v1.1.0/go.mod h1:cTgwzPIzzgDAYoQrMm0EdrjRUBkTqKYp
github.com/vishvananda/netns v0.0.0-20191106174202-0a2b9b5464df/go.mod h1:JP3t17pCcGlemwknint6hfoeCVQrEMVwxRLRjXpq+BU=
github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f h1:p4VB7kIXpOQvVn1ZaTIVp+3vuYAXFe3OJEvjbUYJLaA=
github.com/vishvananda/netns v0.0.0-20210104183010-2eb08e3e575f/go.mod h1:DD4vA1DwXk04H54A1oHXtwZmA0grkVMdPxx/VGLCah0=
github.com/vorot93/golang-signals v0.0.0-20170221070717-d9e83421ce45 h1:hB/hkjwf3BQnZE6Wk3SBwMJz0NqnGdwXoNzHVSYb0N0=
github.com/vorot93/golang-signals v0.0.0-20170221070717-d9e83421ce45/go.mod h1:dfjQkJsG5auteUbnfLIcU72Y/z8tj7DuW9fik8f2Zn0=
github.com/webview/webview v0.0.0-20221218140943-9db4b8c3e9af h1:QO77Qt3ucuL5l8tFIAMXrLaDx4rYI9Nz89x+nrJwYJo=
github.com/webview/webview v0.0.0-20221218140943-9db4b8c3e9af/go.mod h1:rpXAuuHgyEJb6kXcXldlkOjU6y4x+YcASKKXJNUhh0Y=
github.com/webview/webview v0.0.0-20221220082822-77e021440a0f h1:whomWpMJmyN9uHtIjBIE2jmjN8ZccngNh3kCg4G16FA=

3
makefile
View file

@ -27,6 +27,9 @@ $(foreach platf, $(ALL_PLATFORMS), $(addprefix $(platf)-,$(ALL_EXE))):
$(MAKE) $(OUT_DIR)/$(call take,1,$@)-$(call take,2,$@)
GOOS=$(call take,1,$@) GOARCH=$(call take,2,$@) $(BUILD_ENV) ./build -g $(call take,3,$@) -b $(OUT_DIR)/$(call take,1,$@)-$(call take,2,$@)
lint:
golangci-lint run
clean:
help:

230
src/admin/admin.go
View file

@ -5,39 +5,25 @@ import (
"errors"
"fmt"
"net"
"net/http"
"net/url"
"os"
"sort"
"strconv"
"archive/zip"
"strings"
"time"
"gerace.dev/zipfs"
"github.com/RiV-chain/RiV-mesh/src/config"
"github.com/RiV-chain/RiV-mesh/src/core"
"github.com/RiV-chain/RiV-mesh/src/defaults"
)
// TODO: Add authentication
type ServerEvent struct {
event string
data string
}
type AdminSocket struct {
core *core.Core
log core.Logger
listener net.Listener
handlers map[string]handler
done chan struct{}
serverEvents chan ServerEvent
serverEventNextId int
config struct {
core *core.Core
log core.Logger
listener net.Listener
handlers map[string]handler
done chan struct{}
config struct {
listenaddr ListenAddress
}
}
@ -112,7 +98,6 @@ func New(c *core.Core, log core.Logger, opts ...SetupOption) (*AdminSocket, erro
return res, nil
})
a.done = make(chan struct{})
a.serverEvents = make(chan ServerEvent)
go a.listen()
return a, a.core.SetAdmin(a)
}
@ -246,209 +231,6 @@ func (a *AdminSocket) SetupAdminHandlers() {
//_ = a.AddHandler("debug_remoteGetDHT", []string{"key"}, t.proto.getDHTHandler)
}
// Start runs http server
func (a *AdminSocket) StartHttpServer(configFn string, nc *config.NodeConfig) {
if nc.HttpAddress != "none" && nc.HttpAddress != "" && nc.WwwRoot != "none" && nc.WwwRoot != "" {
u, err := url.Parse(nc.HttpAddress)
if err != nil {
a.log.Errorln("An error occurred parsing http address:", err)
return
}
addNoCacheHeaders := func(w http.ResponseWriter) {
w.Header().Add("Cache-Control", "no-cache, no-store, must-revalidate")
w.Header().Add("Pragma", "no-cache")
w.Header().Add("Expires", "0")
}
http.HandleFunc("/api", func(w http.ResponseWriter, r *http.Request) {
fmt.Fprintf(w, "Following methods are allowed: getself, getpeers. litening"+u.Host)
})
http.HandleFunc("/api/self", func(w http.ResponseWriter, r *http.Request) {
addNoCacheHeaders(w)
switch r.Method {
case "GET":
w.Header().Add("Content-Type", "application/json")
req := &GetSelfRequest{}
res := &GetSelfResponse{}
if err := a.getSelfHandler(req, res); err != nil {
http.Error(w, err.Error(), 503)
}
b, err := json.Marshal(res)
if err != nil {
http.Error(w, err.Error(), 503)
}
fmt.Fprint(w, string(b[:]))
default:
http.Error(w, "Method Not Allowed", 405)
}
})
http.HandleFunc("/api/peers", func(w http.ResponseWriter, r *http.Request) {
var handleDelete = func() error {
err := a.core.RemovePeers()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}
return err
}
var handlePost = func() error {
var peers []string
err := json.NewDecoder(r.Body).Decode(&peers)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return err
}
for _, peer := range peers {
if err := a.core.AddPeer(peer, ""); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return err
}
}
if len(configFn) > 0 {
saveHeaders := r.Header["Riv-Save-Config"]
if len(saveHeaders) > 0 && saveHeaders[0] == "true" {
cfg, err := defaults.ReadConfig(configFn)
if err == nil {
cfg.Peers = peers
err := defaults.WriteConfig(configFn, cfg)
if err != nil {
a.log.Errorln("Config file read error:", err)
}
} else {
a.log.Errorln("Config file read error:", err)
}
}
}
return nil
}
addNoCacheHeaders(w)
switch r.Method {
case "GET":
w.Header().Add("Content-Type", "application/json")
req := &GetPeersRequest{}
res := &GetPeersResponse{}
if err := a.getPeersHandler(req, res); err != nil {
http.Error(w, err.Error(), 503)
}
b, err := json.Marshal(res.Peers)
if err != nil {
http.Error(w, err.Error(), 503)
}
fmt.Fprint(w, string(b[:]))
case "POST":
_ = handlePost()
case "PUT":
if handleDelete() == nil {
if handlePost() == nil {
http.Error(w, "No content", http.StatusNoContent)
}
}
case "DELETE":
if handleDelete() == nil {
http.Error(w, "No content", http.StatusNoContent)
}
default:
http.Error(w, "Method Not Allowed", 405)
}
})
http.HandleFunc("/api/ping", func(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case "POST":
peer_list := []string{}
err := json.NewDecoder(r.Body).Decode(&peer_list)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
go a.ping(peer_list)
http.Error(w, "Accepted", http.StatusAccepted)
default:
http.Error(w, "Method Not Allowed", 405)
}
})
http.HandleFunc("/api/sse", func(w http.ResponseWriter, r *http.Request) {
addNoCacheHeaders(w)
switch r.Method {
case "GET":
w.Header().Add("Content-Type", "text/event-stream")
Loop:
for {
select {
case v := <-a.serverEvents:
fmt.Fprintln(w, "id:", a.serverEventNextId)
fmt.Fprintln(w, "event:", v.event)
fmt.Fprintln(w, "data:", v.data)
fmt.Fprintln(w) //end of event
a.serverEventNextId += 1
default:
break Loop
}
}
default:
http.Error(w, "Method Not Allowed", 405)
}
})
var docFs = ""
pakReader, err := zip.OpenReader(nc.WwwRoot)
if err == nil {
defer pakReader.Close()
fs, err := zipfs.NewZipFileSystem(&pakReader.Reader, zipfs.ServeIndexForMissing())
if err == nil {
http.Handle("/", http.FileServer(fs))
docFs = "zipfs"
}
}
if docFs == "" {
var nocache = func(fs http.Handler) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
addNoCacheHeaders(w)
fs.ServeHTTP(w, r)
}
}
http.Handle("/", nocache(http.FileServer(http.Dir(nc.WwwRoot))))
docFs = "local fs"
}
l, e := net.Listen("tcp4", u.Host)
if e != nil {
a.log.Errorf("Http server start error: %s\n", e)
} else {
a.log.Infof("Http server is listening on %s and is supplied from %s %s\n", nc.HttpAddress, docFs, nc.WwwRoot)
}
go func() {
a.log.Errorln(http.Serve(l, nil))
}()
}
}
func (a *AdminSocket) ping(peers []string) {
for _, u := range peers {
go func(u string) {
data, _ := json.Marshal(map[string]string{"peer": u, "value": strconv.FormatInt(check(u), 10)})
a.serverEvents <- ServerEvent{event: "ping", data: string(data)}
}(u)
}
}
func check(peer string) int64 {
u, e := url.Parse(peer)
if e != nil {
return -1
}
t := time.Now()
_, err := net.DialTimeout("tcp", u.Host, 5*time.Second)
if err != nil {
return -1
}
d := time.Since(t)
return d.Milliseconds()
}
// IsStarted returns true if the module has been started.
func (a *AdminSocket) IsStarted() bool {
select {

20
src/core/core.go
View file

@ -13,6 +13,7 @@ import (
iwt "github.com/Arceliar/ironwood/types"
"github.com/Arceliar/phony"
"github.com/gologme/log"
signals "github.com/vorot93/golang-signals"
"github.com/RiV-chain/RiV-mesh/src/version"
)
@ -25,15 +26,16 @@ type Core struct {
// guarantee that it will be covered by the mutex
phony.Inbox
*iwe.PacketConn
ctx context.Context
cancel context.CancelFunc
secret ed25519.PrivateKey
public ed25519.PublicKey
links links
proto protoHandler
log Logger
addPeerTimer *time.Timer
config struct {
ctx context.Context
cancel context.CancelFunc
secret ed25519.PrivateKey
public ed25519.PublicKey
links links
proto protoHandler
log Logger
addPeerTimer *time.Timer
PeersChangedSignal signals.Signal
config struct {
_peers map[Peer]*linkInfo // configurable after startup
_listeners map[ListenAddress]struct{} // configurable after startup
nodeinfo NodeInfo // immutable after startup

4
src/core/link.go
View file

@ -199,6 +199,9 @@ func (intf *link) handler(dial *linkDial) error {
intf.links.core.log.Infof("Connected %s %s: %s, source %s",
dir, strings.ToUpper(intf.info.linkType), remoteStr, localStr)
time.AfterFunc(time.Millisecond*500, func() {
intf.links.core.PeersChangedSignal.Emit(nil)
})
err = intf.links.core.HandleConn(meta.key, intf.conn, intf.options.priority)
switch err {
case io.EOF, net.ErrClosed, nil:
@ -208,6 +211,7 @@ func (intf *link) handler(dial *linkDial) error {
intf.links.core.log.Infof("Disconnected %s %s: %s, source %s; error: %s",
dir, strings.ToUpper(intf.info.linkType), remoteStr, localStr, err)
}
intf.links.core.PeersChangedSignal.Emit(nil)
if !intf.incoming && dial != nil {
// The connection was one that we dialled, so wait a second and try to

317
src/restapi/rest_server.go Normal file
View file

@ -0,0 +1,317 @@
package restapi
import (
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"net"
"net/http"
"net/url"
"sort"
"strconv"
"strings"
"archive/zip"
"time"
"gerace.dev/zipfs"
"github.com/RiV-chain/RiV-mesh/src/core"
"github.com/RiV-chain/RiV-mesh/src/defaults"
"github.com/RiV-chain/RiV-mesh/src/version"
)
type ServerEvent struct {
Event string
Data string
}
type RestServerCfg struct {
Core *core.Core
Log core.Logger
ListenAddress string
WwwRoot string
ConfigFn string
}
type RestServer struct {
RestServerCfg
listenUrl *url.URL
serverEvents chan ServerEvent
serverEventNextId int
docFsType string
}
func NewRestServer(cfg RestServerCfg) (*RestServer, error) {
a := &RestServer{
RestServerCfg: cfg,
serverEvents: make(chan ServerEvent, 10),
serverEventNextId: 0,
}
if cfg.ListenAddress == "none" || cfg.ListenAddress == "" {
return nil, errors.New("listening address isn't configured")
}
var err error
a.listenUrl, err = url.Parse(cfg.ListenAddress)
if err != nil {
return nil, fmt.Errorf("an error occurred parsing http address: %w", err)
}
pakReader, err := zip.OpenReader(cfg.WwwRoot)
if err == nil {
defer pakReader.Close()
fs, err := zipfs.NewZipFileSystem(&pakReader.Reader, zipfs.ServeIndexForMissing())
if err == nil {
http.Handle("/", http.FileServer(fs))
a.docFsType = "zipfs"
}
}
if a.docFsType == "" {
var nocache = func(fs http.Handler) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
addNoCacheHeaders(w)
fs.ServeHTTP(w, r)
}
}
http.Handle("/", nocache(http.FileServer(http.Dir(cfg.WwwRoot))))
a.docFsType = "local fs"
}
http.HandleFunc("/api", a.apiHandler)
http.HandleFunc("/api/self", a.apiSelfHandler)
http.HandleFunc("/api/peers", a.apiPeersHandler)
http.HandleFunc("/api/ping", a.apiPingHandler)
http.HandleFunc("/api/sse", a.apiSseHandler)
var _ = a.Core.PeersChangedSignal.Connect(func(data interface{}) {
b, err := a.prepareGetPeers()
if err != nil {
a.Log.Errorf("get peers failed: %w", err)
return
}
select {
case a.serverEvents <- ServerEvent{Event: "peers", Data: string(b)}:
default:
}
})
return a, nil
}
// Start http server
func (a *RestServer) Serve() error {
l, e := net.Listen("tcp4", a.listenUrl.Host)
if e != nil {
return fmt.Errorf("http server start error: %w", e)
} else {
a.Log.Infof("Http server is listening on %s and is supplied from %s %s\n", a.ListenAddress, a.docFsType, a.WwwRoot)
}
go func() {
err := http.Serve(l, nil)
if err != nil {
a.Log.Errorln(err)
}
}()
return nil
}
func addNoCacheHeaders(w http.ResponseWriter) {
w.Header().Add("Cache-Control", "no-cache, no-store, must-revalidate")
w.Header().Add("Pragma", "no-cache")
w.Header().Add("Expires", "0")
}
func (a *RestServer) apiHandler(w http.ResponseWriter, r *http.Request) {
fmt.Fprintf(w, "Following methods are allowed: GET /api/self, getpeers. litening")
}
func (a *RestServer) apiSelfHandler(w http.ResponseWriter, r *http.Request) {
addNoCacheHeaders(w)
switch r.Method {
case "GET":
w.Header().Add("Content-Type", "application/json")
self := a.Core.GetSelf()
snet := a.Core.Subnet()
var result = map[string]interface{}{
"build_name": a.Core.GetSelf(),
"build_version": version.BuildVersion(),
"key": hex.EncodeToString(self.Key[:]),
"private_key": hex.EncodeToString(self.PrivateKey[:]),
"address": a.Core.Address().String(),
"coords": snet.String(),
"subnet": self.Coords,
}
b, err := json.Marshal(result)
if err != nil {
http.Error(w, err.Error(), http.StatusServiceUnavailable)
return
}
fmt.Fprint(w, string(b[:]))
default:
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
}
}
func (a *RestServer) prepareGetPeers() ([]byte, error) {
peers := a.Core.GetPeers()
response := make([]map[string]interface{}, 0, len(peers))
for _, p := range peers {
addr := a.Core.AddrForKey(p.Key)
response = append(response, map[string]interface{}{
"address": net.IP(addr[:]).String(),
"key": hex.EncodeToString(p.Key),
"port": p.Port,
"priority": uint64(p.Priority), // can't be uint8 thanks to gobind
"coords": p.Coords,
"remote": p.Remote,
"bytes_recvd": p.RXBytes,
"bytes_sent": p.TXBytes,
"uptime": p.Uptime.Seconds(),
"multicast": strings.Contains(p.Remote, "[fe80::"),
})
}
sort.Slice(response, func(i, j int) bool {
if !response[i]["multicast"].(bool) && response[j]["multicast"].(bool) {
return true
}
if response[i]["priority"].(uint64) < response[j]["priority"].(uint64) {
return true
}
return response[i]["port"].(uint64) < response[j]["port"].(uint64)
})
return json.Marshal(response)
}
func (a *RestServer) apiPeersHandler(w http.ResponseWriter, r *http.Request) {
var handleDelete = func() error {
err := a.Core.RemovePeers()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
}
return err
}
var handlePost = func() error {
var peers []string
err := json.NewDecoder(r.Body).Decode(&peers)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return err
}
for _, peer := range peers {
if err := a.Core.AddPeer(peer, ""); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return err
}
}
if len(a.ConfigFn) > 0 {
saveHeaders := r.Header["Riv-Save-Config"]
if len(saveHeaders) > 0 && saveHeaders[0] == "true" {
cfg, err := defaults.ReadConfig(a.ConfigFn)
if err == nil {
cfg.Peers = peers
err := defaults.WriteConfig(a.ConfigFn, cfg)
if err != nil {
a.Log.Errorln("Config file read error:", err)
}
} else {
a.Log.Errorln("Config file read error:", err)
}
}
}
return nil
}
addNoCacheHeaders(w)
switch r.Method {
case "GET":
w.Header().Add("Content-Type", "application/json")
b, err := a.prepareGetPeers()
if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
fmt.Fprint(w, string(b[:]))
case "POST":
_ = handlePost()
case "PUT":
if handleDelete() == nil {
if handlePost() == nil {
http.Error(w, "No content", http.StatusNoContent)
}
}
case "DELETE":
if handleDelete() == nil {
http.Error(w, "No content", http.StatusNoContent)
}
default:
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
}
}
func (a *RestServer) apiPingHandler(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case "POST":
peer_list := []string{}
err := json.NewDecoder(r.Body).Decode(&peer_list)
if err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
go a.ping(peer_list)
http.Error(w, "Accepted", http.StatusAccepted)
default:
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
}
}
func (a *RestServer) apiSseHandler(w http.ResponseWriter, r *http.Request) {
addNoCacheHeaders(w)
switch r.Method {
case "GET":
w.Header().Add("Content-Type", "text/event-stream")
Loop:
for {
select {
case v := <-a.serverEvents:
fmt.Fprintln(w, "id:", a.serverEventNextId)
fmt.Fprintln(w, "event:", v.Event)
fmt.Fprintln(w, "data:", v.Data)
fmt.Fprintln(w) //end of event
a.serverEventNextId += 1
default:
break Loop
}
}
default:
http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
}
}
func (a *RestServer) ping(peers []string) {
for _, u := range peers {
go func(u string) {
data, _ := json.Marshal(map[string]string{"peer": u, "value": strconv.FormatInt(check(u), 10)})
a.serverEvents <- ServerEvent{Event: "ping", Data: string(data)}
}(u)
}
}
func check(peer string) int64 {
u, e := url.Parse(peer)
if e != nil {
return -1
}
t := time.Now()
_, err := net.DialTimeout("tcp", u.Host, 5*time.Second)
if err != nil {
return -1
}
d := time.Since(t)
return d.Milliseconds()
}