<commit_before>// session.go - mixnet client session
// Copyright (C) 2018  <NAME>.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package session

import (
	"encoding/hex"
	"errors"
	"fmt"
	mrand "math/rand"
	"path/filepath"
	"sync"
	"time"

	bolt "github.com/coreos/bbolt"
	"github.com/katzenpost/client/config"
	cConstants "github.com/katzenpost/client/constants"
	"github.com/katzenpost/client/internal/pkiclient"
	"github.com/katzenpost/client/poisson"
	"github.com/katzenpost/client/utils"
	cconstants "github.com/katzenpost/core/constants"
	"github.com/katzenpost/core/crypto/ecdh"
	"github.com/katzenpost/core/crypto/rand"
	"github.com/katzenpost/core/log"
	"github.com/katzenpost/core/pki"
	"github.com/katzenpost/core/sphinx"
	"github.com/katzenpost/core/sphinx/constants"
	sConstants "github.com/katzenpost/core/sphinx/constants"
	cutils "github.com/katzenpost/core/utils"
	"github.com/katzenpost/core/worker"
	"github.com/katzenpost/minclient"
	"gopkg.in/op/go-logging.v1"
)

type Session struct {
	worker.Worker

	cfg       *config.Config
	db        *bolt.DB
	pkiClient pki.Client
	minclient *minclient.Client
	log       *logging.Logger

	fatalErrCh chan error
	haltedCh   chan interface{}
	haltOnce   sync.Once

	// λP, λD and λL Poisson processes
	// as described in "The Loopix Anonymity System".
	pTimer *poisson.PoissonTimer
	dTimer *poisson.PoissonTimer
	lTimer *poisson.PoissonTimer

	linkKey        *ecdh.PrivateKey
	opCh           chan workerOp
	onlineAt       time.Time
	hasPKIDoc      bool
	condGotPKIDoc  *sync.Cond
	condGotConnect *sync.Cond

	egressQueue    EgressQueue
	surbIDMap      map[[sConstants.SURBIDLength]byte]*MessageRef
	messageIDMap   map[[cConstants.MessageIDLength]byte]*MessageRef
	replyNotifyMap map[[cConstants.MessageIDLength]byte]*sync.Mutex
}

// New establishes a session with provider using key.
// This method will block until session is connected to the Provider.
func New(fatalErrCh chan error, logBackend *log.Backend, cfg *config.Config) (*Session, error) {
	var err error

	// create a pkiclient for our own client lookups
	proxyCfg := cfg.UpstreamProxyConfig()
	pkiClient, err := cfg.NonvotingAuthority.New(logBackend, proxyCfg)
	if err != nil {
		return nil, err
	}

	// create a pkiclient for minclient's use
	pkiClient2, err := cfg.NonvotingAuthority.New(logBackend, proxyCfg)
	if err != nil {
		return nil, err
	}
	pkiCacheClient := pkiclient.New(pkiClient2)

	log := logBackend.GetLogger(fmt.Sprintf("%s@%s_c", cfg.Account.User, cfg.Account.Provider))

	s := &Session{
		cfg:        cfg,
		pkiClient:  pkiClient,
		log:        log,
		fatalErrCh: fatalErrCh,
		opCh:       make(chan workerOp),
	}

	// XXX todo: replace all this with persistent data store
	s.surbIDMap = make(map[[sConstants.SURBIDLength]byte]*MessageRef)
	s.messageIDMap = make(map[[cConstants.MessageIDLength]byte]*MessageRef)
	s.replyNotifyMap = make(map[[cConstants.MessageIDLength]byte]*sync.Mutex)
	s.egressQueue = new(Queue)

	// make some synchronised conditions
	s.condGotPKIDoc = sync.NewCond(new(sync.Mutex))
	s.condGotConnect = sync.NewCond(new(sync.Mutex))

	id := cfg.Account.User + "@" + cfg.Account.Provider
	basePath := filepath.Join(cfg.Proxy.DataDir, id)
	if err := cutils.MkDataDir(basePath); err != nil {
		return nil, err
	}

	err = s.loadKeys(basePath)
	if err != nil {
		return nil, err
	}

	// Configure and bring up the minclient instance.
	clientCfg := &minclient.ClientConfig{
		User:                cfg.Account.User,
		Provider:            cfg.Account.Provider,
		ProviderKeyPin:      cfg.Account.ProviderKeyPin,
		LinkKey:             s.linkKey,
		LogBackend:          logBackend,
		PKIClient:           pkiCacheClient,
		OnConnFn:            s.onConnection,
		OnMessageFn:         s.onMessage,
		OnACKFn:             s.onACK,
		OnDocumentFn:        s.onDocument,
		DialContextFn:       proxyCfg.ToDialContext("nonvoting:" + cfg.NonvotingAuthority.PublicKey.String()),
		MessagePollInterval: time.Duration(cfg.Debug.PollingInterval) * time.Second,
		EnableTimeSync:      false, // Be explicit about it.
	}

	s.minclient, err = minclient.New(clientCfg)
	if err != nil {
		return nil, err
	}

	s.Go(s.worker)
	return s, nil
}

func (s *Session) initDatabase(basePath string) error {
	var err error
	s.db, err = bolt.Open(filepath.Join(basePath, "storage.db"), 0600, nil)
	if err != nil {
		return err
	}

	// Initialize (or load) all the buckets.
	err = s.db.Update(func(tx *bolt.Tx) error {

		return nil
	})
	if err != nil {
		s.db.Close()
		s.db = nil
	}
	return err
}

func (s *Session) loadKeys(basePath string) error {
	// Load link key.
	linkPriv := filepath.Join(basePath, "link.private.pem")
	linkPub := filepath.Join(basePath, "link.public.pem")
	var err error
	if s.linkKey, err = ecdh.Load(linkPriv, linkPub, rand.Reader); err != nil {
		s.log.Errorf("Failure to load link keys: %s", err)
		return err
	}
	return nil
}

// GetService returns a randomly selected service
// matching the specified service name
func (s *Session) GetService(serviceName string) (*utils.ServiceDescriptor, error) {
	doc := s.minclient.CurrentDocument()
	if doc == nil {
		return nil, errors.New("pki doc is nil")
	}
	serviceDescriptors := utils.FindServices(serviceName, doc)
	if len(serviceDescriptors) == 0 {
		return nil, errors.New("GetService failure, service not found in pki doc.")
	}
	return &serviceDescriptors[mrand.Intn(len(serviceDescriptors))], nil<commit_msg>Remove all references to boltdb<commit_after>// session.go - mixnet client session
// Copyright (C) 2018  <NAME>.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <http://www.gnu.org/licenses/>.

package session

import (
	"encoding/hex"
	"errors"
	"fmt"
	mrand "math/rand"
	"path/filepath"
	"sync"
	"time"

	"github.com/katzenpost/client/config"
	cConstants "github.com/katzenpost/client/constants"
	"github.com/katzenpost/client/internal/pkiclient"
	"github.com/katzenpost/client/poisson"
	"github.com/katzenpost/client/utils"
	cconstants "github.com/katzenpost/core/constants"
	"github.com/katzenpost/core/crypto/ecdh"
	"github.com/katzenpost/core/crypto/rand"
	"github.com/katzenpost/core/log"
	"github.com/katzenpost/core/pki"
	"github.com/katzenpost/core/sphinx"
	"github.com/katzenpost/core/sphinx/constants"
	sConstants "github.com/katzenpost/core/sphinx/constants"
	cutils "github.com/katzenpost/core/utils"
	"github.com/katzenpost/core/worker"
	"github.com/katzenpost/minclient"
	"gopkg.in/op/go-logging.v1"
)

type Session struct {
	worker.Worker

	cfg       *config.Config
	pkiClient pki.Client
	minclient *minclient.Client
	log       *logging.Logger

	fatalErrCh chan error
	haltedCh   chan interface{}
	haltOnce   sync.Once

	// λP, λD and λL Poisson processes
	// as described in "The Loopix Anonymity System".
	pTimer *poisson.PoissonTimer
	dTimer *poisson.PoissonTimer
	lTimer *poisson.PoissonTimer

	linkKey        *ecdh.PrivateKey
	opCh           chan workerOp
	onlineAt       time.Time
	hasPKIDoc      bool
	condGotPKIDoc  *sync.Cond
	condGotConnect *sync.Cond

	egressQueue    EgressQueue
	surbIDMap      map[[sConstants.SURBIDLength]byte]*MessageRef
	messageIDMap   map[[cConstants.MessageIDLength]byte]*MessageRef
	replyNotifyMap map[[cConstants.MessageIDLength]byte]*sync.Mutex
}

// New establishes a session with provider using key.
// This method will block until session is connected to the Provider.
func New(fatalErrCh chan error, logBackend *log.Backend, cfg *config.Config) (*Session, error) {
	var err error

	// create a pkiclient for our own client lookups
	proxyCfg := cfg.UpstreamProxyConfig()
	pkiClient, err := cfg.NonvotingAuthority.New(logBackend, proxyCfg)
	if err != nil {
		return nil, err
	}

	// create a pkiclient for minclient's use
	pkiClient2, err := cfg.NonvotingAuthority.New(logBackend, proxyCfg)
	if err != nil {
		return nil, err
	}
	pkiCacheClient := pkiclient.New(pkiClient2)

	log := logBackend.GetLogger(fmt.Sprintf("%s@%s_c", cfg.Account.User, cfg.Account.Provider))

	s := &Session{
		cfg:        cfg,
		pkiClient:  pkiClient,
		log:        log,
		fatalErrCh: fatalErrCh,
		opCh:       make(chan workerOp),
	}

	// XXX todo: replace all this with persistent data store
	s.surbIDMap = make(map[[sConstants.SURBIDLength]byte]*MessageRef)
	s.messageIDMap = make(map[[cConstants.MessageIDLength]byte]*MessageRef)
	s.replyNotifyMap = make(map[[cConstants.MessageIDLength]byte]*sync.Mutex)
	s.egressQueue = new(Queue)

	// make some synchronised conditions
	s.condGotPKIDoc = sync.NewCond(new(sync.Mutex))
	s.condGotConnect = sync.NewCond(new(sync.Mutex))

	id := cfg.Account.User + "@" + cfg.Account.Provider
	basePath := filepath.Join(cfg.Proxy.DataDir, id)
	if err := cutils.MkDataDir(basePath); err != nil {
		return nil, err
	}

	err = s.loadKeys(basePath)
	if err != nil {
		return nil, err
	}

	// Configure and bring up the minclient instance.
	clientCfg := &minclient.ClientConfig{
		User:                cfg.Account.User,
		Provider:            cfg.Account.Provider,
		ProviderKeyPin:      cfg.Account.ProviderKeyPin,
		LinkKey:             s.linkKey,
		LogBackend:          logBackend,
		PKIClient:           pkiCacheClient,
		OnConnFn:            s.onConnection,
		OnMessageFn:         s.onMessage,
		OnACKFn:             s.onACK,
		OnDocumentFn:        s.onDocument,
		DialContextFn:       proxyCfg.ToDialContext("nonvoting:" + cfg.NonvotingAuthority.PublicKey.String()),
		MessagePollInterval: time.Duration(cfg.Debug.PollingInterval) * time.Second,
		EnableTimeSync:      false, // Be explicit about it.
	}

	s.minclient, err = minclient.New(clientCfg)
	if err != nil {
		return nil, err
	}

	s.Go(s.worker)
	return s, nil
}

func (s *Session) loadKeys(basePath string) error {
	// Load link key.
	linkPriv := filepath.Join(basePath, "link.private.pem")
	linkPub := filepath.Join(basePath, "link.public.pem")
	var err error
	if s.linkKey, err = ecdh.Load(linkPriv, linkPub, rand.Reader); err != nil {
		s.log.Errorf("Failure to load link keys: %s", err)
		return err
	}
	return nil
}

// GetService returns a randomly selected service
// matching the specified service name
func (s *Session) GetService(serviceName string) (*utils.ServiceDescriptor, error) {
	doc := s.minclient.CurrentDocument()
	if doc == nil {
		return nil, errors.New("pki doc is nil")
	}
	serviceDescriptors := utils.FindServices(serviceName, doc)
	if len(serviceDescriptors) == 0 {
		return nil, errors.New("GetService failure, service not found in pki doc.")
	}
	return &serviceDescriptors[mrand.Intn(len(serviceDescriptors))], nil
{'fraction_non_alphanumeric': 0.1128288349846259, 'fraction_numerical': 0.0018790570550051248, 'mean_word_length': 3.7366504854368934, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2712864', 'n_tokens_mistral': 4062, 'n_tokens_neox': 3956, 'n_words': 999}