cwtch.im
/
cwtch
16
20
Fork 16
Code Issues 10 Pull Requests 2 Releases 163 Wiki Activity

go update
continuous-integration/drone/pr Build is failing Details

This commit is contained in:
Sarah Jamie Lewis 2022-10-03 13:05:36 -07:00
parent 7863ed2aef
commit bb0246b8d9
3 changed files with 817 additions and 817 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.drone.yml
View File

@ -5,7 +5,7 @@ name: linux-test
steps: steps:
- name: fetch - name: fetch
image: golang:1.17.5 image: golang:1.19.1
volumes: volumes:
- name: deps - name: deps
path: /go path: /go

480
app/app.go
View File

@ -1,57 +1,57 @@
package app package app
import ( import (
"cwtch.im/cwtch/app/plugins" "cwtch.im/cwtch/app/plugins"
"cwtch.im/cwtch/event" "cwtch.im/cwtch/event"
"cwtch.im/cwtch/model" "cwtch.im/cwtch/model"
"cwtch.im/cwtch/model/attr" "cwtch.im/cwtch/model/attr"
"cwtch.im/cwtch/model/constants" "cwtch.im/cwtch/model/constants"
"cwtch.im/cwtch/peer" "cwtch.im/cwtch/peer"
"cwtch.im/cwtch/protocol/connections" "cwtch.im/cwtch/protocol/connections"
"cwtch.im/cwtch/storage" "cwtch.im/cwtch/storage"
"git.openprivacy.ca/openprivacy/connectivity" "git.openprivacy.ca/openprivacy/connectivity"
"git.openprivacy.ca/openprivacy/log" "git.openprivacy.ca/openprivacy/log"
"os" "os"
path "path/filepath" path "path/filepath"
"strconv" "strconv"
"sync" "sync"
) )
type application struct { type application struct {
eventBuses map[string]event.Manager eventBuses map[string]event.Manager
directory string directory string
peerLock sync.Mutex peerLock sync.Mutex
peers map[string]peer.CwtchPeer peers map[string]peer.CwtchPeer
acn connectivity.ACN acn connectivity.ACN
plugins sync.Map //map[string] []plugins.Plugin plugins sync.Map //map[string] []plugins.Plugin
engines map[string]connections.Engine engines map[string]connections.Engine
appBus event.Manager appBus event.Manager
appmutex sync.Mutex appmutex sync.Mutex
} }
// Application is a full cwtch peer application. It allows management, usage and storage of multiple peers // Application is a full cwtch peer application. It allows management, usage and storage of multiple peers
type Application interface { type Application interface {
LoadProfiles(password string) LoadProfiles(password string)
CreateTaggedPeer(name string, password string, tag string) CreateTaggedPeer(name string, password string, tag string)
ImportProfile(exportedCwtchFile string, password string) (peer.CwtchPeer, error) ImportProfile(exportedCwtchFile string, password string) (peer.CwtchPeer, error)
DeletePeer(onion string, currentPassword string) DeletePeer(onion string, currentPassword string)
AddPeerPlugin(onion string, pluginID plugins.PluginID) AddPeerPlugin(onion string, pluginID plugins.PluginID)
GetPrimaryBus() event.Manager GetPrimaryBus() event.Manager
GetEventBus(onion string) event.Manager GetEventBus(onion string) event.Manager
QueryACNStatus() QueryACNStatus()
QueryACNVersion() QueryACNVersion()
ActivePeerEngine(onion string, doListen, doPeers, doServers bool) ActivePeerEngine(onion string, doListen, doPeers, doServers bool)
DeactivatePeerEngine(onion string) DeactivatePeerEngine(onion string)
ShutdownPeer(string) ShutdownPeer(string)
Shutdown() Shutdown()
GetPeer(onion string) peer.CwtchPeer GetPeer(onion string) peer.CwtchPeer
ListProfiles() []string ListProfiles() []string
} }
// LoadProfileFn is the function signature for a function in an app that loads a profile // LoadProfileFn is the function signature for a function in an app that loads a profile
@ -59,295 +59,295 @@ type LoadProfileFn func(profile peer.CwtchPeer)
// NewApp creates a new app with some environment awareness and initializes a Tor Manager // NewApp creates a new app with some environment awareness and initializes a Tor Manager
func NewApp(acn connectivity.ACN, appDirectory string) Application { func NewApp(acn connectivity.ACN, appDirectory string) Application {
log.Debugf("NewApp(%v)\n", appDirectory) log.Debugf("NewApp(%v)\n", appDirectory)
os.MkdirAll(path.Join(appDirectory, "profiles"), 0700) os.MkdirAll(path.Join(appDirectory, "profiles"), 0700)
app := &application{engines: make(map[string]connections.Engine), eventBuses: make(map[string]event.Manager), directory: appDirectory, appBus: event.NewEventManager()} app := &application{engines: make(map[string]connections.Engine), eventBuses: make(map[string]event.Manager), directory: appDirectory, appBus: event.NewEventManager()}
app.peers = make(map[string]peer.CwtchPeer) app.peers = make(map[string]peer.CwtchPeer)
app.acn = acn app.acn = acn
statusHandler := app.getACNStatusHandler() statusHandler := app.getACNStatusHandler()
acn.SetStatusCallback(statusHandler) acn.SetStatusCallback(statusHandler)
acn.SetVersionCallback(app.getACNVersionHandler()) acn.SetVersionCallback(app.getACNVersionHandler())
prog, status := acn.GetBootstrapStatus() prog, status := acn.GetBootstrapStatus()
statusHandler(prog, status) statusHandler(prog, status)
return app return app
} }
// ListProfiles returns a map of onions to their profile's Name // ListProfiles returns a map of onions to their profile's Name
func (app *application) ListProfiles() []string { func (app *application) ListProfiles() []string {
var keys []string var keys []string
app.peerLock.Lock() app.peerLock.Lock()
defer app.peerLock.Unlock() defer app.peerLock.Unlock()
for handle := range app.peers { for handle := range app.peers {
keys = append(keys, handle) keys = append(keys, handle)
} }
return keys return keys
} }
// GetPeer returns a cwtchPeer for a given onion address // GetPeer returns a cwtchPeer for a given onion address
func (app *application) GetPeer(onion string) peer.CwtchPeer { func (app *application) GetPeer(onion string) peer.CwtchPeer {
if peer, ok := app.peers[onion]; ok { if peer, ok := app.peers[onion]; ok {
return peer return peer
} }
return nil return nil
} }
func (ap *application) AddPlugin(peerid string, id plugins.PluginID, bus event.Manager, acn connectivity.ACN) { func (ap *application) AddPlugin(peerid string, id plugins.PluginID, bus event.Manager, acn connectivity.ACN) {
if _, exists := ap.plugins.Load(peerid); !exists { if _, exists := ap.plugins.Load(peerid); !exists {
ap.plugins.Store(peerid, []plugins.Plugin{}) ap.plugins.Store(peerid, []plugins.Plugin{})
} }
pluginsinf, _ := ap.plugins.Load(peerid) pluginsinf, _ := ap.plugins.Load(peerid)
peerPlugins := pluginsinf.([]plugins.Plugin) peerPlugins := pluginsinf.([]plugins.Plugin)
newp, err := plugins.Get(id, bus, acn, peerid) newp, err := plugins.Get(id, bus, acn, peerid)
if err == nil { if err == nil {
newp.Start() newp.Start()
peerPlugins = append(peerPlugins, newp) peerPlugins = append(peerPlugins, newp)
log.Debugf("storing plugin for %v %v", peerid, peerPlugins) log.Debugf("storing plugin for %v %v", peerid, peerPlugins)
ap.plugins.Store(peerid, peerPlugins) ap.plugins.Store(peerid, peerPlugins)
} else { } else {
log.Errorf("error adding plugin: %v", err) log.Errorf("error adding plugin: %v", err)
} }
} }
func (app *application) CreateTaggedPeer(name string, password string, tag string) { func (app *application) CreateTaggedPeer(name string, password string, tag string) {
app.appmutex.Lock() app.appmutex.Lock()
defer app.appmutex.Unlock() defer app.appmutex.Unlock()
profileDirectory := path.Join(app.directory, "profiles", model.GenerateRandomID()) profileDirectory := path.Join(app.directory, "profiles", model.GenerateRandomID())
profile, err := peer.CreateEncryptedStorePeer(profileDirectory, name, password) profile, err := peer.CreateEncryptedStorePeer(profileDirectory, name, password)
if err != nil { if err != nil {
log.Errorf("Error Creating Peer: %v", err) log.Errorf("Error Creating Peer: %v", err)
app.appBus.Publish(event.NewEventList(event.PeerError, event.Error, err.Error())) app.appBus.Publish(event.NewEventList(event.PeerError, event.Error, err.Error()))
return return
} }
eventBus := event.NewEventManager() eventBus := event.NewEventManager()
app.eventBuses[profile.GetOnion()] = eventBus app.eventBuses[profile.GetOnion()] = eventBus
profile.Init(app.eventBuses[profile.GetOnion()]) profile.Init(app.eventBuses[profile.GetOnion()])
app.peers[profile.GetOnion()] = profile app.peers[profile.GetOnion()] = profile
if tag != "" { if tag != "" {
profile.SetScopedZonedAttribute(attr.LocalScope, attr.ProfileZone, constants.Tag, tag) profile.SetScopedZonedAttribute(attr.LocalScope, attr.ProfileZone, constants.Tag, tag)
} }
app.appBus.Publish(event.NewEvent(event.NewPeer, map[event.Field]string{event.Identity: profile.GetOnion(), event.Created: event.True})) app.appBus.Publish(event.NewEvent(event.NewPeer, map[event.Field]string{event.Identity: profile.GetOnion(), event.Created: event.True}))
} }
func (app *application) DeletePeer(onion string, password string) { func (app *application) DeletePeer(onion string, password string) {
log.Infof("DeletePeer called on %v\n", onion) log.Infof("DeletePeer called on %v\n", onion)
app.appmutex.Lock() app.appmutex.Lock()
defer app.appmutex.Unlock() defer app.appmutex.Unlock()
if app.peers[onion].CheckPassword(password) { if app.peers[onion].CheckPassword(password) {
app.shutdownPeer(onion) app.shutdownPeer(onion)
app.peers[onion].Delete() app.peers[onion].Delete()
// Shutdown and Remove the Engine // Shutdown and Remove the Engine
log.Debugf("Delete peer for %v Done\n", onion) log.Debugf("Delete peer for %v Done\n", onion)
app.appBus.Publish(event.NewEventList(event.PeerDeleted, event.Identity, onion)) app.appBus.Publish(event.NewEventList(event.PeerDeleted, event.Identity, onion))
return return
} }
app.appBus.Publish(event.NewEventList(event.AppError, event.Error, event.PasswordMatchError, event.Identity, onion)) app.appBus.Publish(event.NewEventList(event.AppError, event.Error, event.PasswordMatchError, event.Identity, onion))
} }
func (app *application) AddPeerPlugin(onion string, pluginID plugins.PluginID) { func (app *application) AddPeerPlugin(onion string, pluginID plugins.PluginID) {
app.AddPlugin(onion, pluginID, app.eventBuses[onion], app.acn) app.AddPlugin(onion, pluginID, app.eventBuses[onion], app.acn)
} }
func (app *application) ImportProfile(exportedCwtchFile string, password string) (peer.CwtchPeer, error) { func (app *application) ImportProfile(exportedCwtchFile string, password string) (peer.CwtchPeer, error) {
profileDirectory := path.Join(app.directory, "profiles") profileDirectory := path.Join(app.directory, "profiles")
profile, err := peer.ImportProfile(exportedCwtchFile, profileDirectory, password) profile, err := peer.ImportProfile(exportedCwtchFile, profileDirectory, password)
if profile != nil || err == nil { if profile != nil || err == nil {
app.installProfile(profile) app.installProfile(profile)
} }
return profile, err return profile, err
} }
// LoadProfiles takes a password and attempts to load any profiles it can from storage with it and create Peers for them // LoadProfiles takes a password and attempts to load any profiles it can from storage with it and create Peers for them
func (app *application) LoadProfiles(password string) { func (app *application) LoadProfiles(password string) {
count := 0 count := 0
migrating := false migrating := false
files, err := os.ReadDir(path.Join(app.directory, "profiles")) files, err := os.ReadDir(path.Join(app.directory, "profiles"))
if err != nil { if err != nil {
log.Errorf("error: cannot read profiles directory: %v", err) log.Errorf("error: cannot read profiles directory: %v", err)
return return
} }
for _, file := range files { for _, file := range files {
// Attempt to load an encrypted database // Attempt to load an encrypted database
profileDirectory := path.Join(app.directory, "profiles", file.Name()) profileDirectory := path.Join(app.directory, "profiles", file.Name())
profile, err := peer.FromEncryptedDatabase(profileDirectory, password) profile, err := peer.FromEncryptedDatabase(profileDirectory, password)
loaded := false loaded := false
if err == nil { if err == nil {
// return the load the profile... // return the load the profile...
log.Infof("loading profile from new-type storage database...") log.Infof("loading profile from new-type storage database...")
loaded = app.installProfile(profile) loaded = app.installProfile(profile)
} else { // On failure attempt to load a legacy profile } else { // On failure attempt to load a legacy profile
profileStore, err := storage.LoadProfileWriterStore(profileDirectory, password) profileStore, err := storage.LoadProfileWriterStore(profileDirectory, password)
if err != nil { if err != nil {
continue continue
} }
log.Infof("found legacy profile. importing to new database structure...") log.Infof("found legacy profile. importing to new database structure...")
legacyProfile := profileStore.GetProfileCopy(true) legacyProfile := profileStore.GetProfileCopy(true)
if !migrating { if !migrating {
migrating = true migrating = true
app.appBus.Publish(event.NewEventList(event.StartingStorageMiragtion)) app.appBus.Publish(event.NewEventList(event.StartingStorageMiragtion))
} }
cps, err := peer.CreateEncryptedStore(profileDirectory, password) cps, err := peer.CreateEncryptedStore(profileDirectory, password)
if err != nil { if err != nil {
log.Errorf("error creating encrypted store: %v", err) log.Errorf("error creating encrypted store: %v", err)
} }
profile := peer.ImportLegacyProfile(legacyProfile, cps) profile := peer.ImportLegacyProfile(legacyProfile, cps)
loaded = app.installProfile(profile) loaded = app.installProfile(profile)
} }
if loaded { if loaded {
count++ count++
} }
} }
if count == 0 { if count == 0 {
message := event.NewEventList(event.AppError, event.Error, event.AppErrLoaded0) message := event.NewEventList(event.AppError, event.Error, event.AppErrLoaded0)
app.appBus.Publish(message) app.appBus.Publish(message)
} }
if migrating { if migrating {
app.appBus.Publish(event.NewEventList(event.DoneStorageMigration)) app.appBus.Publish(event.NewEventList(event.DoneStorageMigration))
} }
} }
// installProfile takes a profile and if it isn't loaded in the app, installs it and returns true // installProfile takes a profile and if it isn't loaded in the app, installs it and returns true
func (app *application) installProfile(profile peer.CwtchPeer) bool { func (app *application) installProfile(profile peer.CwtchPeer) bool {
app.appmutex.Lock() app.appmutex.Lock()
defer app.appmutex.Unlock() defer app.appmutex.Unlock()
// Only attempt to finalize the profile if we don't have one loaded... // Only attempt to finalize the profile if we don't have one loaded...
if app.peers[profile.GetOnion()] == nil { if app.peers[profile.GetOnion()] == nil {
eventBus := event.NewEventManager() eventBus := event.NewEventManager()
app.eventBuses[profile.GetOnion()] = eventBus app.eventBuses[profile.GetOnion()] = eventBus
profile.Init(app.eventBuses[profile.GetOnion()]) profile.Init(app.eventBuses[profile.GetOnion()])
app.peers[profile.GetOnion()] = profile app.peers[profile.GetOnion()] = profile
app.appBus.Publish(event.NewEvent(event.NewPeer, map[event.Field]string{event.Identity: profile.GetOnion(), event.Created: event.False})) app.appBus.Publish(event.NewEvent(event.NewPeer, map[event.Field]string{event.Identity: profile.GetOnion(), event.Created: event.False}))
return true return true
} }
// Otherwise shutdown the connections // Otherwise shutdown the connections
profile.Shutdown() profile.Shutdown()
return false return false
} }
// ActivePeerEngine creates a peer engine for use with an ACN, should be called once the underlying ACN is online // ActivePeerEngine creates a peer engine for use with an ACN, should be called once the underlying ACN is online
func (app *application) ActivePeerEngine(onion string, doListen, doPeers, doServers bool) { func (app *application) ActivePeerEngine(onion string, doListen, doPeers, doServers bool) {
profile := app.GetPeer(onion) profile := app.GetPeer(onion)
if profile != nil { if profile != nil {
app.engines[profile.GetOnion()], _ = profile.GenerateProtocolEngine(app.acn, app.eventBuses[profile.GetOnion()]) app.engines[profile.GetOnion()], _ = profile.GenerateProtocolEngine(app.acn, app.eventBuses[profile.GetOnion()])
if doListen { if doListen {
profile.Listen() profile.Listen()
} }
if doPeers { if doPeers {
profile.StartPeersConnections() profile.StartPeersConnections()
} }
if doServers { if doServers {
profile.StartServerConnections() profile.StartServerConnections()
} }
} }
} }
// DeactivatePeerEngine shutsdown and cleans up a peer engine, should be called when an underlying ACN goes offline // DeactivatePeerEngine shutsdown and cleans up a peer engine, should be called when an underlying ACN goes offline
func (app *application) DeactivatePeerEngine(onion string) { func (app *application) DeactivatePeerEngine(onion string) {
if engine, exists := app.engines[onion]; exists { if engine, exists := app.engines[onion]; exists {
engine.Shutdown() engine.Shutdown()
delete(app.engines, onion) delete(app.engines, onion)
} }
} }
// GetPrimaryBus returns the bus the Application uses for events that aren't peer specific // GetPrimaryBus returns the bus the Application uses for events that aren't peer specific
func (app *application) GetPrimaryBus() event.Manager { func (app *application) GetPrimaryBus() event.Manager {
return app.appBus return app.appBus
} }
// GetEventBus returns a cwtchPeer's event bus // GetEventBus returns a cwtchPeer's event bus
func (app *application) GetEventBus(onion string) event.Manager { func (app *application) GetEventBus(onion string) event.Manager {
if manager, ok := app.eventBuses[onion]; ok { if manager, ok := app.eventBuses[onion]; ok {
return manager return manager
} }
return nil return nil
} }
func (app *application) getACNStatusHandler() func(int, string) { func (app *application) getACNStatusHandler() func(int, string) {
return func(progress int, status string) { return func(progress int, status string) {
progStr := strconv.Itoa(progress) progStr := strconv.Itoa(progress)
app.appmutex.Lock() app.appmutex.Lock()
app.appBus.Publish(event.NewEventList(event.ACNStatus, event.Progress, progStr, event.Status, status)) app.appBus.Publish(event.NewEventList(event.ACNStatus, event.Progress, progStr, event.Status, status))
for _, bus := range app.eventBuses { for _, bus := range app.eventBuses {
bus.Publish(event.NewEventList(event.ACNStatus, event.Progress, progStr, event.Status, status)) bus.Publish(event.NewEventList(event.ACNStatus, event.Progress, progStr, event.Status, status))
} }
app.appmutex.Unlock() app.appmutex.Unlock()
} }
} }
func (app *application) getACNVersionHandler() func(string) { func (app *application) getACNVersionHandler() func(string) {
return func(version string) { return func(version string) {
app.appmutex.Lock() app.appmutex.Lock()
defer app.appmutex.Unlock() defer app.appmutex.Unlock()
app.appBus.Publish(event.NewEventList(event.ACNVersion, event.Data, version)) app.appBus.Publish(event.NewEventList(event.ACNVersion, event.Data, version))
} }
} }
func (app *application) QueryACNStatus() { func (app *application) QueryACNStatus() {
prog, status := app.acn.GetBootstrapStatus() prog, status := app.acn.GetBootstrapStatus()
app.getACNStatusHandler()(prog, status) app.getACNStatusHandler()(prog, status)
} }
func (app *application) QueryACNVersion() { func (app *application) QueryACNVersion() {
version := app.acn.GetVersion() version := app.acn.GetVersion()
app.appBus.Publish(event.NewEventList(event.ACNVersion, event.Data, version)) app.appBus.Publish(event.NewEventList(event.ACNVersion, event.Data, version))
} }
// ShutdownPeer shuts down a peer and removes it from the app's management // ShutdownPeer shuts down a peer and removes it from the app's management
func (app *application) ShutdownPeer(onion string) { func (app *application) ShutdownPeer(onion string) {
app.appmutex.Lock() app.appmutex.Lock()
defer app.appmutex.Unlock() defer app.appmutex.Unlock()
app.shutdownPeer(onion) app.shutdownPeer(onion)
} }
// shutdownPeer mutex unlocked helper shutdown peer // shutdownPeer mutex unlocked helper shutdown peer
func (app *application) shutdownPeer(onion string) { func (app *application) shutdownPeer(onion string) {
app.eventBuses[onion].Publish(event.NewEventList(event.ShutdownPeer, event.Identity, onion)) app.eventBuses[onion].Publish(event.NewEventList(event.ShutdownPeer, event.Identity, onion))
app.eventBuses[onion].Shutdown() app.eventBuses[onion].Shutdown()
delete(app.eventBuses, onion) delete(app.eventBuses, onion)
app.peers[onion].Shutdown() app.peers[onion].Shutdown()
delete(app.peers, onion) delete(app.peers, onion)
if _, ok := app.engines[onion]; ok { if _, ok := app.engines[onion]; ok {
app.engines[onion].Shutdown() app.engines[onion].Shutdown()
delete(app.engines, onion) delete(app.engines, onion)
} }
log.Debugf("shutting down plugins for %v", onion) log.Debugf("shutting down plugins for %v", onion)
pluginsI, ok := app.plugins.Load(onion) pluginsI, ok := app.plugins.Load(onion)
if ok { if ok {
plugins := pluginsI.([]plugins.Plugin) plugins := pluginsI.([]plugins.Plugin)
for _, plugin := range plugins { for _, plugin := range plugins {
log.Debugf("shutting down plugin: %v", plugin) log.Debugf("shutting down plugin: %v", plugin)
plugin.Shutdown() plugin.Shutdown()
} }
} }
app.plugins.Delete(onion) app.plugins.Delete(onion)
} }
// Shutdown shutsdown all peers of an app // Shutdown shutsdown all peers of an app
func (app *application) Shutdown() { func (app *application) Shutdown() {
app.appmutex.Lock() app.appmutex.Lock()
defer app.appmutex.Unlock() defer app.appmutex.Unlock()
for id := range app.peers { for id := range app.peers {
log.Debugf("Shutting Down Peer %v", id) log.Debugf("Shutting Down Peer %v", id)
app.shutdownPeer(id) app.shutdownPeer(id)
} }
log.Debugf("Shutting Down App") log.Debugf("Shutting Down App")
app.appBus.Shutdown() app.appBus.Shutdown()
log.Debugf("Shut Down Complete") log.Debugf("Shut Down Complete")
} }

1152
protocol/connections/engine.go
View File

File diff suppressed because it is too large Load Diff