Brian tested manually

Merged in ui_permissions (pull request #82)

Ui permissions

Approved-by: Eric Harding <eric.harding@bluetradesys.com>

.vscode/launch.json

@@ -42,7 +42,18 @@
          "stopAtEntry": false,
          "externalConsole": true,
          "requireExactSource": false
-      },        
+      },     
+      {
+         "name": "Client Tests",
+         "type": "clr",
+         "request": "launch",
+         "program": "${workspaceRoot}/build/Debug/bin/net471/sparkcmd.exe",
+         "args": ["test","test_filter"],
+         "cwd": "${workspaceRoot}",
+         "stopAtEntry": false,
+         "externalConsole": true,
+         "requireExactSource": false
+      },           
 		{
             "name": "Client Attach",
             "type": "clr",

src/client/spark/Admin/AdminModel.cs

@@ -11,18 +11,22 @@ namespace mbts.spark {
    [Export]
    class AdminModel : IDisposable {
       readonly string _user_name;
+      public string user_name => _user_name;
+      
       readonly BehaviorSubject<IReadOnlyDictionary<uint, string>>
          _accounts_subject = new BehaviorSubject<IReadOnlyDictionary<uint, string>>(new Dictionary<uint, string>()),
-         _users_subject = new BehaviorSubject<IReadOnlyDictionary<uint, string>>(new Dictionary<uint, string>());
+         _users_subject = new BehaviorSubject<IReadOnlyDictionary<uint, string>>(new Dictionary<uint, string>());               
       IReadOnlyDictionary<uint, string>
          _account_map = new Dictionary<uint, string>(),
          _user_map = new Dictionary<uint, string>();
-      IDisposable _dispose;
+      readonly BehaviorSubject<AdminModelWrapper> _wrapper_subject = new BehaviorSubject<AdminModelWrapper>(null);         
+      IDisposable _dispose;      
 
       [ImportingConstructor]
       AdminModel(AdminService service, [Import("username")]string user_name) {
          _user_name = user_name;
          _dispose = service.subscribei_admin_data().Select(update => {
+            _wrapper_subject.OnNext(new AdminModelWrapper(update.all.FirstOrDefault(), _user_name));
             IEnumerable<BTSUserImmutable> active_users =
                from admin_data in update.all
                from bts_user in admin_data.Users
@@ -52,9 +56,10 @@ namespace mbts.spark {
       }
 
       public IObservable<IReadOnlyDictionary<uint, string>> on_users => _users_subject.AsObservable();
-      public IObservable<IReadOnlyDictionary<uint, string>> on_accounts => _accounts_subject.AsObservable();
+      public IObservable<IReadOnlyDictionary<uint, string>> on_accounts => _accounts_subject.AsObservable();      
       public IObservable<IReadOnlyList<UserTuple>> on_user_tuples => on_users.Select(d => d.Select(kv => new UserTuple(kv.Key, kv.Value)).ToList());
       public IObservable<IReadOnlyList<AccountTuple>> on_account_tuples => on_accounts.Select(d => d.Select(kv => new AccountTuple(kv.Key, kv.Value)).ToList());
+      public IObservable<AdminModelWrapper> on_admin_data => _wrapper_subject.AsObservable();
 
       public UserTuple? get_current_user() {
          foreach (var kv in _user_map) {

src/client/spark/Admin/AdminModelWrapper.cs

@@ -0,0 +1,65 @@
+using bts;
+using bts.api.admin;
+using System;
+using System.Collections.Generic;
+using System.Linq;
+
+namespace mbts.spark 
+{
+   class AdminModelWrapper
+   {
+      readonly AdminDataImmutable? _admin_data;
+      readonly string _user_name;
+      readonly List<BTSAccountImmutable> _allowed_accounts;
+      readonly Dictionary<string, string> _allowed_account_dict = new Dictionary<string, string>();
+      public AdminModelWrapper(AdminDataImmutable? admin_data = null, string user_name = null)
+      {
+         _admin_data = admin_data;
+         _user_name = user_name;
+         _allowed_accounts = search_admin(u => u.Name == _user_name).Select(x => x.Item1).ToList();
+         foreach(var val in _allowed_accounts)
+            _allowed_account_dict[val.Name] = val.Name;
+      }
+      public IEnumerable<Tuple<BTSAccountImmutable, TradingPermissionUnitImmutable>> get_trading_permissions(UInt32 user_id)
+      {
+         return search_admin(u => u.Id == user_id);
+      }         
+      public IEnumerable<BTSAccountImmutable> viewable_accounts()
+      {
+         return _allowed_accounts;
+      }
+      public bool can_view_account(string account)
+      {
+         return _allowed_account_dict.ContainsKey(account);
+      }
+
+      IEnumerable<Tuple<BTSAccountImmutable, TradingPermissionUnitImmutable>> search_admin(Func<BTSUserImmutable, bool> predicate)
+      {
+         var rv = new List<Tuple<BTSAccountImmutable, TradingPermissionUnitImmutable>>();
+         if (_admin_data.HasValue)
+         {
+            var users = _admin_data.Value.Users.Where(predicate);
+            if (users != null && users.Count() == 1)
+            {
+               var user = users.ElementAt(0);
+               if (user.HasUserData && user.UserData.HasAccountPermissions)
+               {
+                  foreach (var kv in user.UserData.AccountPermissions.AccountPermissions)
+                  {
+                     var account_id = kv.Key;
+                     var accounts = _admin_data.Value.Accounts.Where(a => a.Id == account_id);
+                     if (accounts != null && accounts.Count() == 1)
+                     {
+                        var account = accounts.ElementAt(0);
+                        var tp = kv.Value.TradingPermission;
+                        var rv_val = new TradingPermissionUnitImmutable(tp);
+                        rv.Add(Tuple.Create(account, rv_val));
+                     }
+                  }
+               }
+            }
+         }
+         return rv;
+      }
+   }
+}

src/client/spark/Admin/PermissionFilters.cs

@@ -0,0 +1,66 @@
+using bts;
+using bts.api;
+using bts.api.core;
+using bts.api.orderman;
+using bts.api.position;
+using bts.spark;
+using mbts.collections;
+using mbts.spark;
+using mbts.spark.utilities;
+using mbts.utils;
+using System;
+using System.Collections.Generic;
+using System.Collections.Immutable;
+using System.Collections.Specialized;
+using System.Linq;
+using System.Reactive.Concurrency;
+using System.Reactive.Linq;
+
+namespace bts.api.core
+{
+
+   static class PermissionFilters
+   {
+      public static IObservable<Update<PositionUnitImmutable, int>> PermissionFilter(this IObservable<Update<PositionUnitImmutable, int>> source, AdminModel am)
+      {
+         var pred = Observable.Create<Func<PositionUnitImmutable, bool>>( observer =>
+         {
+            return am.on_admin_data.Subscribe(x =>
+            {
+               Func<PositionUnitImmutable, bool> filter = (z => x.can_view_account(z.Account));
+               observer.OnNext(filter);
+            });
+         });
+
+         return source.Filter(pred);
+      }
+
+      public static IObservable<Update<OrderRecordImmutable, uint>> PermissionFilter(this IObservable<Update<OrderRecordImmutable, uint>> source, AdminModel am)
+      {
+         var pred = Observable.Create<Func<OrderRecordImmutable, bool>>(observer =>
+        {
+           return am.on_admin_data.Subscribe(x =>
+           {
+              Func<OrderRecordImmutable, bool> filter = (z => x.can_view_account(z.account()));
+              observer.OnNext(filter);
+           });
+        });
+
+         return source.Filter(pred);
+      }
+
+      public static IObservable<Update<FillImmutable, string>> PermissionFilter(this IObservable<Update<FillImmutable, string>> source, AdminModel am)
+      {
+         var pred = Observable.Create<Func<FillImmutable, bool>>(observer =>
+        {
+           return am.on_admin_data.Subscribe(x =>
+           {
+              Func<FillImmutable, bool> filter = (z => x.can_view_account(z.Account));
+              observer.OnNext(filter);
+           });
+        });
+
+         return source.Filter(pred);
+      }
+   }
+}

src/client/spark/Orders/OrdersModel.cs

@@ -32,7 +32,7 @@ namespace mbts.spark {
       IDisposable _dispose;
 
       [ImportingConstructor]
-      private OrdersModel(IBus bus, OrderFactory factory, OrderService service, PositionService position_service, SpreaderModel spreader, ErrorReporter error) {
+      private OrdersModel(IBus bus, OrderFactory factory, OrderService service, PositionService position_service, SpreaderModel spreader, ErrorReporter error, AdminModel admin_model) {
          _bus = bus;
          _factory = factory;
          _service = service;
@@ -41,15 +41,15 @@ namespace mbts.spark {
          _error = error;
          var scheduler = ThreadPoolScheduler.Instance;
          var interval = TimeSpan.FromMilliseconds(100);
-         on_orders = _service.subscribei_orders().Naggle(interval, scheduler).Table().Cache();
-         on_positions = _position_service.subscribei_position_data().Naggle(interval, scheduler).Table().Cache();
+         on_orders = _service.subscribei_orders().Naggle(interval, scheduler).PermissionFilter(admin_model).Table().Cache();
+         on_positions = _position_service.subscribei_position_data().Naggle(interval, scheduler).PermissionFilter(admin_model).Table().Cache();
          working_orders = 
             on_orders.Scan(
                ImmutableDictionary<uint,OrderRecordImmutable>.Empty, 
                (workers, table) => _update_working_orders(workers, table))
                .Replay(1).RefCount();
          on_order_tuples = OrderRecordTupleSource.create_observable(bus.subscribe(SparkTags.SecuritiesTag), on_orders);
-         on_fills = _service.subscribei_fills().Naggle(interval, scheduler).Table().Cache();
+         on_fills = _service.subscribei_fills().Naggle(interval, scheduler).PermissionFilter(admin_model).Table().Cache();
          _activity = new OrderActivitySummarizer(bus, on_orders);
          on_fill_tuples = FillTupleSource.create_observable(bus.subscribe(SparkTags.SecuritiesTag), on_orders, on_fills);
          _security_orders_demuxer = new TableUpdateDemuxer<OrderRecordImmutable, uint, uint>(o => o.Key, o => o.security_key());

src/client/spark_api/api/Update.cs

@@ -129,6 +129,7 @@ namespace bts.api.core {
       public IEnumerable<T> removed { get; }
 
       public TKey key(T t) { return _key(t); }
+      internal Func<T, TKey> key_extractor => _key;
 
 #if DEBUG
       public override string ToString() {
@@ -347,5 +348,72 @@ namespace bts.api.core {
          return source.Select(t => new LegacyTableUpdate<T, TKey>(t));
       }
 
+      public static IObservable<Update<T, TKey>> Filter<T, TKey>(this IObservable<Update<T, TKey>> source, IObservable<Func<T, bool>> predicate) where TKey : IEquatable<TKey>
+      {
+         return Observable.Create<Update<T, TKey>>(observer =>
+         {
+            Update<T, TKey> unfiltered_cache = null, filtered_cache = null;
+            Func<T, bool> pred_cache = null;
+            Log.DataFlow.diag($"Create Filter {typeof(T).Name}");
+
+            return source.CombineLatest(predicate, (upd, pred) => (upd, pred)).Subscribe( tup =>
+            {
+               bool pred_changed = !object.ReferenceEquals(tup.pred, pred_cache) && (pred_cache != null);
+               pred_cache = tup.pred;
+               unfiltered_cache = tup.upd;
+               if (pred_changed && filtered_cache != null)
+               {
+                  //clear out existing
+                  var clear = new Update<T, TKey>(
+                     (IReadOnlyCollection<T>)Array.Empty<T>(), 
+                     Enumerable.Empty<T>(),
+                     Enumerable.Empty<Change<T>>(),
+                     filtered_cache.all, unfiltered_cache.key_extractor);
+                  observer.OnNext(clear);
+
+                  var refilter = unfiltered_cache.all.Where(tup.pred).ToList();
+                  var readd = new Update<T, TKey>(
+                     refilter, refilter, 
+                     Enumerable.Empty<Change<T>>(),
+                     Enumerable.Empty<T>(), unfiltered_cache.key_extractor);
+                  observer.OnNext(readd);
+                  filtered_cache = readd;
+               }
+               else
+               {
+                  IReadOnlyCollection<T> all; IEnumerable<T> added, removed; IEnumerable<Change<T>> changed;
+                  apply_predicate(tup.pred, tup.upd, out all, out added, out changed, out removed);
+                  var next = new Update<T, TKey>(all, added, changed, removed, tup.upd.key_extractor);
+                  observer.OnNext(next);
+                  filtered_cache = next;
+               }
+            }, 
+            () => observer.OnCompleted());
+         });
+      }
+
+      private static void apply_predicate<T, TKey>(Func<T, bool> predicate, Update<T, TKey> cache, 
+                                                   out IReadOnlyCollection<T> all, out IEnumerable<T> added, 
+                                                   out IEnumerable<Change<T>> changed, out IEnumerable<T> removed) where TKey : IEquatable<TKey>
+      {                 
+         var adds = new List<T>();
+         adds.AddRange(cache.added.Where(predicate));
+         var rems = new List<T>();
+         rems.AddRange(cache.removed.Where(predicate));
+         var diffs = new List<Change<T>>();
+         foreach (var change in cache.changes)
+         {
+            if(!predicate(change.Old) && predicate(change.New))
+               adds.Add(change.New);
+            else if(predicate(change.Old) && !predicate(change.New))
+               rems.Add(change.Old);
+            else if(predicate(change.Old) && predicate(change.New))
+               diffs.Add(change);
+         }
+         all = cache.all.Where(predicate).ToList();
+         added = adds;
+         changed = diffs;
+         removed = rems;
+      }
    }
 }

src/client/spark_test/api/ApiUpdateTests.cs

@@ -6,6 +6,7 @@ using System.Reactive.Linq;
 using System.Threading.Tasks;
 using System.Linq;
 using System.Reactive.Concurrency;
+using System.Reactive.Subjects;
 
 namespace mbts.api.Test {
    [TestGroup]
@@ -104,6 +105,121 @@ namespace mbts.api.Test {
 
       }
 
+      void validate_filter(IObservable<RawUpdate<Item>> raw, IObservable<Update<Item, int>> filtered, Func<Item, bool> pred)
+      {
+         var total = raw.ToEnumerable().Count();
+
+         // Validate that update is right at each point
+         for (var count = 1; count < total; ++count)
+         {
+            var orig = raw.Take(count).reconstitute(id).ToEnumerable().Last();            
+            var next = filtered.Take(count).ToEnumerable().Last();
+            Assert.That(next.all.All(pred));
+            Assert.That(next.added.All(pred));
+            Assert.That(next.removed.All(pred));
+            Assert.That(next.changed.All(pred));            
+         }
+
+         // Do it again with TableUpdate
+         for (var count = 1; count < total; ++count)
+         {
+            var orig = raw.Take(count).reconstitute(id).Table().ToEnumerable().Last();
+            var next = filtered.Take(count).ToEnumerable().Last();
+            Assert.That(next.added.All(pred));
+            Assert.That(next.removed.All(pred));
+            Assert.That(next.changed.All(pred));
+         }
+      }
+
+
+      void validate_filter_sub(IObservable<RawUpdate<Item>> raw, IObservable<Update<Item, int>> filtered, BehaviorSubject<Func<Item, bool>> pred)
+      {
+         raw.reconstitute(x => x.id).CombineLatest(filtered, (r, f) => Tuple.Create(r, f)).Subscribe(tup =>
+         {
+            Assert.That(tup.Item2.all.All(pred.Value));
+            Assert.That(tup.Item2.added.All(pred.Value));
+            Assert.That(!tup.Item2.removed.Any(pred.Value)); 
+            Assert.That(tup.Item2.changed.All(pred.Value));
+
+            Dictionary<int, Item> rawd = tup.Item1.added.ToDictionary(x => x.id);
+            Dictionary<int, Item> fd = tup.Item2.added.ToDictionary(x => x.id);
+            foreach (var kv in rawd)
+            {
+               if(pred.Value(kv.Value))
+                  Assert.That(fd.ContainsKey(kv.Key));
+               else 
+                  Assert.That(!fd.ContainsKey(kv.Key));
+            }
+         });
+      }
+
+
+      [Test]
+      public void test_filter()
+      {
+         var raw = Observable.Create<RawUpdate<Item>>(observer =>
+         {
+            RawUpdate<Item> rup = new RawUpdate<Item>();
+            for (int x = 0; x < 50; ++x)
+            {
+               rup.Clear();
+               var item = _item_pool[x];
+               rup.added_items.Add(item);
+               rup.added_keys.Add(item.id);
+               observer.OnNext(rup);
+            }
+            var rand = new Random();
+            for (int r = 0; r < rand.Next(4, 30); r += rand.Next(1, 2))
+            {
+               rup.Clear();
+               rup.removed_keys.Add(_item_pool[r].id);
+               observer.OnNext(rup);
+            }
+            rup.Clear();
+            for (int c = 31; c < rand.Next(32, 50); c += rand.Next(1, 2))
+            {               
+               var ch = new Item(_item_pool[c].id, _item_pool[c].name + "_change");
+               rup.changed_items.Add(ch);
+               rup.changed_keys.Add(ch.id);               
+            }
+            observer.OnNext(rup);
+
+            observer.OnCompleted(); 
+            return Disposable.empty;
+         });
+
+         Func<Item, bool> evens = (x => x.id % 2 == 0);
+         Func<Item, bool> odds = (x => x.id % 2 == 1);
+         var predeven = Observable.Create<Func<Item, bool>>(observer =>
+         {            
+            observer.OnNext(evens);            
+            observer.OnCompleted();
+            return Disposable.empty;
+         });
+
+         var filtered = raw.reconstitute(x=> x.id).Filter(predeven);
+         validate_filter(raw, filtered, evens);
+
+         var predodd = Observable.Create<Func<Item, bool>>(observer =>
+         {
+            observer.OnNext(odds);
+            observer.OnCompleted();
+            return Disposable.empty;
+         });
+
+         filtered = raw.reconstitute(x => x.id).Filter(predodd);
+         validate_filter(raw, filtered, odds);
+
+
+
+         var sub= new BehaviorSubject<Func<Item, bool>>(evens);
+         filtered = raw.reconstitute(x => x.id).Filter(sub);
+         validate_filter_sub(raw, filtered, sub);
+
+         sub.OnNext(odds);
+         sub.OnCompleted();
+      }      
+
       [Test]
       public async Task test_add() {
          // Add 1 item 10x

testing/manual/admin/user_account_permissions.txt

@@ -0,0 +1,14 @@
+This test verifies that users can only see trades that are in accounts for which they're permissioned
+
+1. Start a client and the admin gui
+
+2. For your user, add permissions to see all accounts in the system in the admin gui
+
+3. Enter some orders in the client for those accounts (you may have to add limits, etc), make some orders fill, cancel some orders
+
+4. Now close the client, and in the admin UI pick another user and remove trading access to the account you just traded into.   Start the client again with the user you just de-permissioned.  You should not be able to see any of the trades you just made
+
+5. Next re-permission your current user for the accounts with trades in them and the should show up in the UI.
+
+6. Start another client with -multiple with the original user and make sure you can still see all trades.
+

tools/runsparktest.cmd

@@ -0,0 +1,4 @@
+@echo off
+pushd %~dp0\..\src\client\sparkcmd
+dotnet watch run test %1 -t
+popd