Make some operations exempt from app permissions. (#3851)

## Motivation

Chains with restricted block operations can't process e.g. new events
because system operations are not allowed.

## Proposal

Allow `Process*Epoch` and `UpdateStreams`.
Enforce that `UpdateStreams` actually has new events.

## Test Plan

A test was extended.

## Release Plan

- These changes should be backported to the latest `testnet` branch,
then
    - be released in a new SDK,
    - be released in a validator hotfix.

## Links

- Fixes #3849.
- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

Author: afck

linera-chain/src/chain.rs

@@ -1100,6 +1100,10 @@ where
             app_permissions.mandatory_applications.iter().cloned(),
         );
         for operation in &block.operations {
+            if operation.is_exempt_from_permissions() {
+                mandatory.clear();
+                continue;
+            }
             ensure!(
                 app_permissions.can_execute_operations(&operation.application_id()),
                 ChainError::AuthorizedApplications(

linera-core/src/unit_tests/client_tests.rs

@@ -9,9 +9,9 @@ mod wasm;
 use assert_matches::assert_matches;
 use futures::StreamExt;
 use linera_base::{
-    crypto::AccountSecretKey,
+    crypto::{AccountSecretKey, CryptoHash},
     data_types::*,
-    identifiers::{Account, AccountOwner, ChainId, MessageId},
+    identifiers::{Account, AccountOwner, ApplicationId, ChainId, MessageId},
     ownership::{ChainOwnership, TimeoutConfig},
 };
 use linera_chain::{
@@ -1161,7 +1161,7 @@ where
     assert_eq!(user.epoch().await.unwrap(), Epoch::from(1));
 
     // Create a new committee.
-    let committee = Committee::new(validators, ResourceControlPolicy::only_fuel());
+    let committee = Committee::new(validators.clone(), ResourceControlPolicy::only_fuel());
     admin.stage_new_committee(committee).await.unwrap();
     assert_eq!(admin.next_block_height(), BlockHeight::from(5));
     assert!(admin.pending_proposal().is_none());
@@ -1236,6 +1236,22 @@ where
     admin.process_inbox().await.unwrap();
     // Transfer goes through and the previous one as well thanks to block chaining.
     assert_eq!(admin.local_balance().await.unwrap(), Amount::from_tokens(3));
+
+    user.change_application_permissions(ApplicationPermissions::new_single(ApplicationId::new(
+        CryptoHash::test_hash("foo"),
+    )))
+    .await?;
+
+    let committee = Committee::new(validators, ResourceControlPolicy::default());
+    admin.stage_new_committee(committee).await.unwrap();
+    assert_eq!(admin.epoch().await.unwrap(), Epoch::from(3));
+
+    // Despite the restrictive application permissions, some system operations are still allowed,
+    // and the user chain can migrate to the new epoch.
+    user.synchronize_from_validators().await?;
+    user.process_inbox().await?;
+    assert_eq!(user.epoch().await.unwrap(), Epoch::from(3));
+
     Ok(())
 }
 

linera-execution/src/lib.rs

@@ -333,6 +333,8 @@ pub enum ExecutionError {
     InternalError(&'static str),
     #[error("UpdateStreams contains an unknown event")]
     EventNotFound(EventId),
+    #[error("UpdateStreams is outdated")]
+    OutdatedUpdateStreams,
 }
 
 impl From<ViewError> for ExecutionError {
@@ -1211,6 +1213,19 @@ impl Operation {
             _ => vec![],
         }
     }
+
+    /// Returns whether this operation is allowed regardless of application permissions.
+    pub fn is_exempt_from_permissions(&self) -> bool {
+        let Operation::System(system_op) = self else {
+            return false;
+        };
+        matches!(
+            **system_op,
+            SystemOperation::ProcessNewEpoch(_)
+                | SystemOperation::ProcessRemovedEpoch(_)
+                | SystemOperation::UpdateStreams(_)
+        )
+    }
 }
 
 impl From<SystemMessage> for Message {

linera-execution/src/system.rs

@@ -519,9 +519,10 @@ where
                         .event_subscriptions
                         .get_mut_or_default(&(chain_id, stream_id.clone()))
                         .await?;
-                    if subscriptions.next_index >= next_index {
-                        continue;
-                    }
+                    ensure!(
+                        subscriptions.next_index < next_index,
+                        ExecutionError::OutdatedUpdateStreams
+                    );
                     for application_id in &subscriptions.applications {
                         txn_tracker.add_stream_to_process(
                             *application_id,