iota_source_validation_service/
lib.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::{
6    collections::BTreeMap,
7    ffi::OsString,
8    fmt, fs,
9    net::TcpListener,
10    path::{Path, PathBuf},
11    process::Command,
12    sync::{Arc, RwLock},
13};
14
15use anyhow::{anyhow, bail};
16use axum::{
17    Extension, Json, Router,
18    extract::{Query, State},
19    middleware::{self, Next},
20    response::{IntoResponse, Response},
21    routing::get,
22};
23use hyper::{
24    HeaderMap, StatusCode,
25    http::{HeaderName, HeaderValue, Method},
26};
27use iota_metrics::RegistryService;
28use iota_move::manage_package::resolve_lock_file_path;
29use iota_move_build::{BuildConfig, IotaPackageHooks, ProtocolBuildConfig, implicit_deps};
30use iota_package_management::system_package_versions::latest_system_packages;
31use iota_sdk::{IotaClientBuilder, rpc_types::IotaTransactionBlockEffects};
32use iota_sdk_types::ObjectId;
33use iota_source_validation::{BytecodeSourceVerifier, ValidationMode};
34use move_core_types::account_address::AccountAddress;
35use move_package::{BuildConfig as MoveBuildConfig, LintFlag};
36use move_symbol_pool::Symbol;
37use prometheus_filtered::{IntCounter, MetricLevel, Registry, register_int_counter_with_registry};
38use serde::{Deserialize, Serialize};
39use tokio::sync::oneshot::Sender;
40use tower::ServiceBuilder;
41use tracing::{debug, info};
42use url::Url;
43
44pub const HOST_PORT_ENV: &str = "HOST_PORT";
45pub const IOTA_SOURCE_VALIDATION_VERSION_HEADER: &str = "x-iota-source-validation-version";
46pub const IOTA_SOURCE_VALIDATION_VERSION: &str = "0.1";
47
48pub const MAINNET_URL: &str = "https://api.mainnet.iota.cafe:443";
49pub const TESTNET_URL: &str = "https://api.testnet.iota.cafe:443";
50pub const DEVNET_URL: &str = "https://api.devnet.iota.cafe:443";
51pub const LOCALNET_URL: &str = "http://127.0.0.1:9000";
52
53pub const METRICS_ROUTE: &str = "/metrics";
54pub const METRICS_HOST_PORT: &str = "0.0.0.0:9184";
55
56pub fn host_port() -> String {
57    match option_env!("HOST_PORT") {
58        Some(v) => v.to_string(),
59        None => String::from("0.0.0.0:8000"),
60    }
61}
62
63#[derive(Clone, Deserialize, Debug)]
64pub struct Config {
65    pub packages: Vec<PackageSource>,
66}
67
68#[derive(Clone, Deserialize, Debug)]
69#[serde(tag = "source", content = "values")]
70pub enum PackageSource {
71    Repository(RepositorySource),
72    Directory(DirectorySource),
73}
74
75#[derive(Clone, Deserialize, Debug)]
76pub struct RepositorySource {
77    pub repository: String,
78    pub network: Option<Network>,
79    pub branches: Vec<Branch>,
80}
81
82#[derive(Clone, Deserialize, Debug)]
83pub struct Branch {
84    pub branch: String,
85    pub paths: Vec<Package>,
86}
87
88#[derive(Clone, Deserialize, Debug)]
89pub struct DirectorySource {
90    pub paths: Vec<Package>,
91    pub network: Option<Network>,
92}
93
94#[derive(Clone, Deserialize, Debug)]
95pub struct Package {
96    pub path: String,
97    /// Optional object ID to watch for upgrades. For framework packages, this
98    /// is an address like 0x2. For non-framework packages this is an
99    /// upgrade cap (possibly wrapped).
100    pub watch: Option<ObjectId>,
101}
102
103#[derive(Clone, Serialize, Debug)]
104pub struct SourceInfo {
105    pub path: PathBuf,
106    #[serde(skip_serializing_if = "Option::is_none")]
107    // Is Some when content is hydrated from disk.
108    pub source: Option<String>,
109}
110
111#[derive(Eq, PartialEq, Clone, Default, Serialize, Deserialize, Debug, Ord, PartialOrd)]
112#[serde(rename_all = "lowercase")]
113pub enum Network {
114    #[default]
115    #[serde(alias = "Mainnet")]
116    Mainnet,
117    #[serde(alias = "Testnet")]
118    Testnet,
119    #[serde(alias = "Devnet")]
120    Devnet,
121    #[serde(alias = "Localnet")]
122    Localnet,
123}
124
125impl fmt::Display for Network {
126    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
127        write!(
128            f,
129            "{}",
130            match self {
131                Network::Mainnet => "mainnet",
132                Network::Testnet => "testnet",
133                Network::Devnet => "devnet",
134                Network::Localnet => "localnet",
135            }
136        )
137    }
138}
139
140/// Map module name to verified source info.
141pub type SourceLookup = BTreeMap<Symbol, SourceInfo>;
142/// Map addresses to module names and sources.
143pub type AddressLookup = BTreeMap<AccountAddress, SourceLookup>;
144/// Top-level lookup that maps network to sources for corresponding on-chain
145/// networks.
146pub type NetworkLookup = BTreeMap<Network, AddressLookup>;
147
148pub async fn verify_package(
149    network: &Network,
150    package_path: impl AsRef<Path>,
151) -> anyhow::Result<(Network, AddressLookup)> {
152    move_package::package_hooks::register_package_hooks(Box::new(IotaPackageHooks));
153    // TODO(rvantonder): use config RPC URL instead of hardcoded URLs
154    let network_url = match network {
155        Network::Mainnet => MAINNET_URL,
156        Network::Testnet => TESTNET_URL,
157        Network::Devnet => DEVNET_URL,
158        Network::Localnet => LOCALNET_URL,
159    };
160    let client = IotaClientBuilder::default().build(network_url).await?;
161    let chain_id = client.read_api().get_chain_identifier().await?;
162    let protocol_config = client.read_api().get_protocol_config(None).await?;
163    let mut config =
164        resolve_lock_file_path(MoveBuildConfig::default(), Some(package_path.as_ref()))?;
165    config.lint_flag = LintFlag::LEVEL_NONE;
166    config.silence_warnings = true;
167    config.implicit_dependencies = implicit_deps(latest_system_packages());
168    let build_config = BuildConfig {
169        config,
170        run_bytecode_verifier: false, // no need to run verifier if code is on-chain
171        print_diags_to_stderr: false,
172        chain_id: Some(chain_id),
173        protocol_build_config: ProtocolBuildConfig::from(&protocol_config),
174    };
175    let compiled_package = build_config.build(package_path.as_ref())?;
176
177    BytecodeSourceVerifier::new(client.read_api())
178        .verify(&compiled_package, ValidationMode::root())
179        .await
180        .map_err(|e| anyhow!("Network {network}: {e}"))?;
181
182    let mut address_map = AddressLookup::new();
183    let address = compiled_package
184        .published_at
185        .as_ref()
186        .map(|id| AccountAddress::new(id.into_bytes()))
187        .map_err(|_| anyhow!("could not resolve published-at field in package manifest"))?;
188    info!("verifying {} at {address}", package_path.as_ref().display());
189    for v in &compiled_package.package.root_compiled_units {
190        let path = v.source_path.to_path_buf();
191        let source = Some(fs::read_to_string(path.as_path())?);
192        let name = v.unit.name;
193        if let Some(existing) = address_map.get_mut(&address) {
194            existing.insert(name, SourceInfo { path, source });
195        } else {
196            let mut source_map = SourceLookup::new();
197            source_map.insert(name, SourceInfo { path, source });
198            address_map.insert(address, source_map);
199        }
200    }
201    Ok((network.clone(), address_map))
202}
203
204pub fn parse_config(config_path: impl AsRef<Path>) -> anyhow::Result<Config> {
205    let contents = fs::read_to_string(config_path)?;
206    Ok(toml::from_str(&contents)?)
207}
208
209pub fn repo_name_from_url(url: &str) -> anyhow::Result<String> {
210    let repo_url = Url::parse(url)?;
211    let mut components = repo_url
212        .path_segments()
213        .ok_or_else(|| anyhow!("Could not discover repository path in url {url}"))?;
214    let repo_name = components
215        .next_back()
216        .ok_or_else(|| anyhow!("Could not discover repository name in url {url}"))?;
217    Ok(repo_name.to_string())
218}
219
220#[derive(Debug)]
221/// Represents a sequence of git commands to clone a repository and sparsely
222/// checkout Move packages within.
223pub struct CloneCommand {
224    /// git args
225    args: Vec<Vec<OsString>>,
226    /// report repository url in error messages
227    repo_url: String,
228}
229
230impl CloneCommand {
231    pub fn new(p: &RepositorySource, b: &Branch, dest: &Path) -> anyhow::Result<CloneCommand> {
232        let repo_name = repo_name_from_url(&p.repository)?;
233        let network = p.network.clone().unwrap_or_default().to_string();
234        let sanitized_branch = b.branch.replace('/', "__");
235        let dest = dest
236            .join(network)
237            .join(format!("{repo_name}__{sanitized_branch}"))
238            .into_os_string();
239
240        macro_rules! ostr {
241            ($arg:expr) => {
242                OsString::from($arg)
243            };
244        }
245
246        let mut args = vec![];
247        // Args to clone empty repository
248        let cmd_args: Vec<OsString> = vec![
249            ostr!("clone"),
250            ostr!("--no-checkout"),
251            ostr!("--depth=1"), // implies --single-branch
252            ostr!("--filter=tree:0"),
253            ostr!(format!("--branch={}", b.branch)),
254            ostr!(&p.repository),
255            ostr!(dest.clone()),
256        ];
257        args.push(cmd_args);
258
259        // Args to sparse checkout the package set
260        let mut cmd_args: Vec<OsString> = vec![
261            ostr!("-C"),
262            dest.clone(),
263            ostr!("sparse-checkout"),
264            ostr!("set"),
265            ostr!("--no-cone"),
266        ];
267        let path_args: Vec<OsString> = b
268            .paths
269            .iter()
270            .map(|p| OsString::from(p.path.clone()))
271            .collect();
272        cmd_args.extend_from_slice(&path_args);
273        args.push(cmd_args);
274
275        // Args to checkout the default branch.
276        let cmd_args: Vec<OsString> = vec![ostr!("-C"), dest, ostr!("checkout")];
277        args.push(cmd_args);
278
279        Ok(Self {
280            args,
281            repo_url: p.repository.clone(),
282        })
283    }
284
285    pub async fn run(&self) -> anyhow::Result<()> {
286        for args in &self.args {
287            let result = Command::new("git").args(args).output().map_err(|_| {
288                anyhow!(
289                    "Error cloning {} with command `git {:#?}`",
290                    self.repo_url,
291                    args
292                )
293            })?;
294            if !result.status.success() {
295                bail!(
296                    "Nonzero exit status when cloning {} with command `git {:#?}`. \
297		     Stderr: {}",
298                    self.repo_url,
299                    args,
300                    String::from_utf8_lossy(&result.stderr)
301                )
302            }
303        }
304        Ok(())
305    }
306}
307
308/// Clones repositories and checks out packages as per `config` at the directory
309/// `dir`.
310pub async fn clone_repositories(repos: Vec<&RepositorySource>, dir: &Path) -> anyhow::Result<()> {
311    let mut tasks = vec![];
312    for p in &repos {
313        for b in &p.branches {
314            let command = CloneCommand::new(p, b, dir)?;
315            info!(
316                "cloning {}:{} to {}",
317                &p.repository,
318                &b.branch,
319                dir.display()
320            );
321            let t = tokio::spawn(async move { command.run().await });
322            tasks.push(t);
323        }
324    }
325
326    for t in tasks {
327        t.await.unwrap()?;
328    }
329    Ok(())
330}
331
332pub async fn initialize(
333    config: &Config,
334    dir: &Path,
335) -> anyhow::Result<(NetworkLookup, NetworkLookup)> {
336    let mut repos = vec![];
337    for s in &config.packages {
338        match s {
339            PackageSource::Repository(r) => repos.push(r),
340            PackageSource::Directory(_) => (), // skip cloning
341        }
342    }
343    clone_repositories(repos, dir).await?;
344    let sources = verify_packages(config, dir).await?;
345    let sources_list = sources_list(&sources).await;
346    Ok((sources, sources_list))
347}
348
349pub async fn sources_list(sources: &NetworkLookup) -> NetworkLookup {
350    let mut sources_list = NetworkLookup::new();
351    for (network, addresses) in sources {
352        let mut address_map = AddressLookup::new();
353        for (address, symbols) in addresses {
354            let mut symbol_map = SourceLookup::new();
355            for (symbol, source_info) in symbols {
356                symbol_map.insert(
357                    *symbol,
358                    SourceInfo {
359                        path: source_info.path.file_name().unwrap().into(),
360                        source: None,
361                    },
362                );
363            }
364            address_map.insert(*address, symbol_map);
365        }
366        sources_list.insert(network.clone(), address_map);
367    }
368    sources_list
369}
370
371pub async fn verify_packages(config: &Config, dir: &Path) -> anyhow::Result<NetworkLookup> {
372    let mut tasks = vec![];
373    for p in &config.packages {
374        match p {
375            PackageSource::Repository(r) => {
376                let repo_name = repo_name_from_url(&r.repository)?;
377                let network_name = r.network.clone().unwrap_or_default().to_string();
378                for b in &r.branches {
379                    for p in &b.paths {
380                        let sanitized_branch = b.branch.replace('/', "__");
381                        let package_path = dir
382                            .join(network_name.clone())
383                            .join(format!("{repo_name}__{sanitized_branch}"))
384                            .join(p.path.clone())
385                            .clone();
386                        let network = r.network.clone().unwrap_or_default();
387                        let t =
388                            tokio::spawn(
389                                async move { verify_package(&network, package_path).await },
390                            );
391                        tasks.push(t)
392                    }
393                }
394            }
395            PackageSource::Directory(packages_dir) => {
396                for p in &packages_dir.paths {
397                    let package_path = PathBuf::from(p.path.clone());
398                    let network = packages_dir.network.clone().unwrap_or_default();
399                    let t =
400                        tokio::spawn(async move { verify_package(&network, package_path).await });
401                    tasks.push(t)
402                }
403            }
404        }
405    }
406
407    let mut mainnet_lookup = AddressLookup::new();
408    let mut testnet_lookup = AddressLookup::new();
409    let mut devnet_lookup = AddressLookup::new();
410    let mut localnet_lookup = AddressLookup::new();
411    for t in tasks {
412        let (network, new_lookup) = t.await.unwrap()?;
413        match network {
414            Network::Mainnet => mainnet_lookup.extend(new_lookup),
415            Network::Testnet => testnet_lookup.extend(new_lookup),
416            Network::Devnet => devnet_lookup.extend(new_lookup),
417            Network::Localnet => localnet_lookup.extend(new_lookup),
418        }
419    }
420    let mut lookup = NetworkLookup::new();
421    lookup.insert(Network::Mainnet, mainnet_lookup);
422    lookup.insert(Network::Testnet, testnet_lookup);
423    lookup.insert(Network::Devnet, devnet_lookup);
424    lookup.insert(Network::Localnet, localnet_lookup);
425    Ok(lookup)
426}
427
428// A thread that monitors on-chain transactions for package upgrades. `config`
429// specifies which packages to watch. `app_state` contains the map of sources
430// returned by the server. In particular, `watch_for_upgrades` invalidates
431// (i.e., clears) the sources returned by the serve when we observe a package
432// upgrade, so that we do not falsely report outdated sources for a package.
433// Pass an optional `channel` to observe the upgrade transaction(s).
434// The `channel` parameter exists for testing.
435pub async fn watch_for_upgrades(
436    _packages: Vec<PackageSource>,
437    _app_state: Arc<RwLock<AppState>>,
438    _network: Network,
439    _channel: Option<Sender<IotaTransactionBlockEffects>>,
440) -> anyhow::Result<()> {
441    Err(anyhow!(
442        "Fatal: JsonRPC Subscriptions no longer supported. Reimplement without using subscriptions."
443    ))
444}
445
446pub struct AppState {
447    pub sources: NetworkLookup,
448    pub metrics: Option<SourceServiceMetrics>,
449    pub sources_list: NetworkLookup,
450}
451
452pub async fn serve(app_state: Arc<RwLock<AppState>>) -> anyhow::Result<()> {
453    let app = Router::new()
454        .route("/api", get(api_route))
455        .route("/api/list", get(list_route))
456        .layer(
457            ServiceBuilder::new()
458                .layer(
459                    tower_http::cors::CorsLayer::new()
460                        .allow_methods([Method::GET])
461                        .allow_origin(tower_http::cors::Any),
462                )
463                .layer(middleware::from_fn(check_version_header)),
464        )
465        .with_state(app_state);
466    let listener = TcpListener::bind(host_port())?;
467    listener.set_nonblocking(true).unwrap();
468    let listener = tokio::net::TcpListener::from_std(listener)?;
469    axum::serve(listener, app).await?;
470    Ok(())
471}
472
473#[derive(Deserialize)]
474pub struct Request {
475    #[serde(default)]
476    network: Network,
477    address: String,
478    module: String,
479}
480
481#[derive(Serialize, Deserialize)]
482pub struct SourceResponse {
483    pub source: String,
484}
485
486#[derive(Serialize, Deserialize)]
487pub struct ErrorResponse {
488    pub error: String,
489}
490
491async fn api_route(
492    State(app_state): State<Arc<RwLock<AppState>>>,
493    Query(Request {
494        network,
495        address,
496        module,
497    }): Query<Request>,
498) -> impl IntoResponse {
499    debug!("request network={network}&address={address}&module={module}");
500    let symbol = Symbol::from(module);
501    let Ok(address) = AccountAddress::from_hex_literal(&address) else {
502        let error = format!("Invalid hex address {address}");
503        return (
504            StatusCode::BAD_REQUEST,
505            Json(ErrorResponse { error }).into_response(),
506        );
507    };
508
509    let app_state = app_state.read().unwrap();
510    if let Some(metrics) = &app_state.metrics {
511        metrics.total_requests_received.inc();
512    }
513    let source_result = app_state
514        .sources
515        .get(&network)
516        .and_then(|n| n.get(&address))
517        .and_then(|a| a.get(&symbol));
518    if let Some(SourceInfo {
519        source: Some(source),
520        ..
521    }) = source_result
522    {
523        (
524            StatusCode::OK,
525            Json(SourceResponse {
526                source: source.to_owned(),
527            })
528            .into_response(),
529        )
530    } else {
531        (
532            StatusCode::NOT_FOUND,
533            Json(ErrorResponse {
534                error: format!(
535                    "No source found for {symbol} at address {address} on network {network}"
536                ),
537            })
538            .into_response(),
539        )
540    }
541}
542
543async fn check_version_header(
544    headers: HeaderMap,
545    req: hyper::Request<axum::body::Body>,
546    next: Next,
547) -> Response {
548    let version = headers
549        .get(IOTA_SOURCE_VALIDATION_VERSION_HEADER)
550        .as_ref()
551        .and_then(|h| h.to_str().ok())
552        .map(|s| s.to_string());
553
554    match version {
555        Some(v) if v != IOTA_SOURCE_VALIDATION_VERSION => {
556            let error = format!(
557                "Unsupported version '{v}' specified in header \
558		 {IOTA_SOURCE_VALIDATION_VERSION_HEADER}"
559            );
560            let mut headers = HeaderMap::new();
561            headers.insert(
562                HeaderName::from_static(IOTA_SOURCE_VALIDATION_VERSION_HEADER),
563                HeaderValue::from_static(IOTA_SOURCE_VALIDATION_VERSION),
564            );
565            return (
566                StatusCode::BAD_REQUEST,
567                headers,
568                Json(ErrorResponse { error }).into_response(),
569            )
570                .into_response();
571        }
572        _ => (),
573    }
574    let mut response = next.run(req).await;
575    response.headers_mut().insert(
576        HeaderName::from_static(IOTA_SOURCE_VALIDATION_VERSION_HEADER),
577        HeaderValue::from_static(IOTA_SOURCE_VALIDATION_VERSION),
578    );
579    response
580}
581
582async fn list_route(State(app_state): State<Arc<RwLock<AppState>>>) -> impl IntoResponse {
583    let app_state = app_state.read().unwrap();
584    (
585        StatusCode::OK,
586        Json(app_state.sources_list.clone()).into_response(),
587    )
588}
589
590pub struct SourceServiceMetrics {
591    pub total_requests_received: IntCounter,
592}
593
594impl SourceServiceMetrics {
595    pub fn new(registry: &Registry) -> Self {
596        Self {
597            total_requests_received: register_int_counter_with_registry!(
598                "total_requests",
599                "Total number of requests received by Source Service",
600                registry;
601                MetricLevel::Info
602            )
603            .unwrap(),
604        }
605    }
606}
607
608pub fn start_prometheus_server(listener: TcpListener) -> RegistryService {
609    let registry = Registry::new();
610
611    let registry_service = RegistryService::new(registry);
612
613    let app = Router::new()
614        .route(METRICS_ROUTE, get(iota_metrics::metrics))
615        .layer(Extension(registry_service.clone()));
616
617    tokio::spawn(async move {
618        listener.set_nonblocking(true).unwrap();
619        let listener = tokio::net::TcpListener::from_std(listener).unwrap();
620        axum::serve(listener, app).await.unwrap();
621    });
622
623    registry_service
624}
iota_source_validation_service/lib.rs

iota_source_validation_service/
lib.rs
