tensamin/omega
Watch
6
0
Fork
You've already forked omega
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

[COMPLETE] MTP-MIGRATION [Some errors to be found]

Browse source
This commit is contained in:
Alex Emmet 2026-07-04 02:41:01 +02:00
commit fb4facb8c4
4 changed files with 69 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

38
src/server/api.rs
View file

@ -1,6 +1,6 @@
use crate::load_keyring;
use crate::sql::sql;
use crate::sql::sql::{get_by_user_id, get_omikron_by_id};
use crate::sql::sql::{get_by_user_id, get_iota_by_id, get_omikron_by_id};
use crate::sql::user_online_tracker::get_iota_primary_omikron_connection;
use crate::transport::omikron_manager::get_random_omikron;
use crate::util::file_util::get_directory;
@ -64,12 +64,13 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
["api", "get", "omikron"] => {
if let Ok(omikron_conn) = get_random_omikron().await {
if let Some(id) = omikron_conn.get_omikron_id().await {
if let Ok((public_key, ip_address)) = sql::get_omikron_by_id(id).await {
if let Ok((public_key, ip_address, port)) = sql::get_omikron_by_id(id).await {
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["id"] = id.into();
res["public_key"] = public_key.to_base64().into();
res["ip_address"] = ip_address.into();
res["port"] = port.into();
(StatusCode::OK, res.dump())
} else {
@ -99,20 +100,22 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
let mut res = JsonValue::new_object();
res["status"] = "error_bad_request".into();
(StatusCode::BAD_REQUEST, res.dump())
} else if let Ok((public_key, ip_address)) = get_omikron_by_id(id).await {
} else if let Ok((public_key, ip_address, port)) = get_omikron_by_id(id).await {
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["id"] = id.into();
res["public_key"] = public_key.to_base64().into();
res["ip_address"] = ip_address.into();
res["port"] = port.into();
(StatusCode::OK, res.dump())
} else if let Some(omikron_id) = get_iota_primary_omikron_connection(id) {
if let Ok((public_key, ip_address)) = get_omikron_by_id(omikron_id).await {
if let Ok((public_key, ip_address, port)) = get_omikron_by_id(omikron_id).await {
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["id"] = omikron_id.into();
res["public_key"] = public_key.to_base64().into();
res["ip_address"] = ip_address.into();
res["port"] = port.into();
(StatusCode::OK, res.dump())
} else {
let mut res = JsonValue::new_object();
@ -122,12 +125,14 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
} else if let Ok((_, iota_id, _, _, _, _, _, _, _, _, _, _)) = get_by_user_id(id).await
{
if let Some(omikron_id) = get_iota_primary_omikron_connection(iota_id) {
if let Ok((public_key, ip_address)) = get_omikron_by_id(omikron_id).await {
if let Ok((public_key, ip_address, port)) = get_omikron_by_id(omikron_id).await
{
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["id"] = omikron_id.into();
res["public_key"] = public_key.to_base64().into();
res["ip_address"] = ip_address.into();
res["port"] = port.into();
(StatusCode::OK, res.dump())
} else {
let mut res = JsonValue::new_object();
@ -146,6 +151,29 @@ pub async fn handle(path: &str, body_string: Option<String>) -> HttpResponse {
}
}
// ==================================================
// GET IOTA BY ID
// ==================================================
["api", "get", "iota", id] => {
let id: i64 = id.parse().unwrap_or(0);
if id == 0 {
let mut res = JsonValue::new_object();
res["status"] = "error_bad_request".into();
(StatusCode::BAD_REQUEST, res.dump())
} else if let Ok((id, public_key)) = get_iota_by_id(id).await {
let mut res = JsonValue::new_object();
res["status"] = "success".into();
res["iota_id"] = id.into();
res["public_key"] = public_key.to_base64().into();
(StatusCode::OK, res.dump())
} else {
let mut res = JsonValue::new_object();
res["status"] = "error_not_found".into();
(StatusCode::NOT_FOUND, res.dump())
}
}
// ==================================================
// GET ID BY USERNAME
// ==================================================

2
src/server/server.rs
View file

@ -28,7 +28,7 @@ pub async fn start(port: u16) -> anyhow::Result<()> {
.with_no_client_auth()
.with_single_cert(cert_chain, key)?;
config.alpn_protocols = vec![b"h2".to_vec(), b"hmtp/1.1".to_vec()];
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
let bind_addr = std::env::var("BIND_ADDRESS").unwrap_or_else(|_| "0.0.0.0".to_string());
let addr = format!("{}:{}", bind_addr, port);

24
src/sql/sql.rs
View file

@ -85,11 +85,19 @@ pub async fn initialize_db() -> Result<(), sqlx::Error> {
id BIGINT NOT NULL PRIMARY KEY,
public_key BLOB NOT NULL,
location VARCHAR(255) NOT NULL COLLATE utf8mb4_bin,
ip_address VARCHAR(255) NOT NULL COLLATE utf8mb4_bin
ip_address VARCHAR(255) NOT NULL COLLATE utf8mb4_bin,
port INT(11) NOT NULL DEFAULT 959
)",
)
.execute(&pool)
.await;
// Retrofits `port` onto omikrons tables created before this column existed;
// `CREATE TABLE IF NOT EXISTS` above is a no-op against an already-existing table.
let _ = sqlx::query(
"ALTER TABLE omikrons ADD COLUMN IF NOT EXISTS port INT(11) NOT NULL DEFAULT 959",
)
.execute(&pool)
.await;
let _ = sqlx::query(
"CREATE TABLE IF NOT EXISTS
notifications (
@ -684,7 +692,7 @@ pub async fn delete_iota(id: i64) -> Result<(), sqlx::Error> {
// OMIKRONS
// ==========================================================================================
pub async fn get_omikron_by_id(id: i64) -> Result<(PublicKeyBundle, String), sqlx::Error> {
pub async fn get_omikron_by_id(id: i64) -> Result<(PublicKeyBundle, String, u16), sqlx::Error> {
let pool = {
let db_lock = SQL_DB.read().await;
db_lock
@ -693,18 +701,22 @@ pub async fn get_omikron_by_id(id: i64) -> Result<(PublicKeyBundle, String), sql
.expect("Database pool not initialized")
};
let row = sqlx::query_as::<_, (Vec<u8>, Vec<u8>)>(
"SELECT public_key, ip_address FROM omikrons WHERE id = ?",
let row = sqlx::query_as::<_, (Vec<u8>, Vec<u8>, i32)>(
"SELECT public_key, ip_address, port FROM omikrons WHERE id = ?",
)
.bind(id)
.fetch_optional(&pool)
.await?;
match row {
Some((public_key, ip_address)) => {
Some((public_key, ip_address, port)) => {
let bundle = PublicKeyBundle::from_bytes(&public_key)
.map_err(|e| sqlx::Error::Decode(Box::new(e)))?;
Ok((bundle, String::from_utf8_lossy(&ip_address).to_string()))
Ok((
bundle,
String::from_utf8_lossy(&ip_address).to_string(),
port as u16,
))
}
_ => Err(sqlx::Error::RowNotFound),
}

22
src/transport/omikron_connection.rs
View file

@ -1051,7 +1051,7 @@ pub async fn get_by_omikron_id(
sql::get_omikron_by_id(omikron_id as i64)
.await
.ok()
.map(|(bundle, _ip_address)| bundle)
.map(|(bundle, _ip_address, _port)| bundle)
}
pub async fn complete_register(_pub_key: PublicKeyBundle, _description: Option<String>) -> u64 {
0
@ -1094,11 +1094,23 @@ pub async fn start(port: u16) -> Result<(), Box<dyn std::error::Error>> {
let mut host: Host = Host::new(host_config).await?;
log!("OmikronServer listening on port {}", port);
while let Ok(Some(mut connection)) = host.accept().await {
loop {
let mut conn = match host.accept().await {
Ok(Some(conn)) => conn,
Ok(None) => break,
Err(e) => {
// A single omikron's failed/aborted handshake (bad auth, a
// probe, a mid-handshake disconnect) must not take down the
// whole listener - only that connection attempt is lost.
log_err!(0, PrintType::Omega, "Rejected omikron connection: {}", e);
continue;
}
};
let omikron_connection = OmikronConnection::new(conn.sender, conn.client_id);
tokio::spawn(async move {
let conn = OmikronConnection::new(connection.sender, connection.client_id);
omikron_manager::add_omikron(conn.clone()).await;
conn.handle(&mut connection.receiver).await;
omikron_manager::add_omikron(omikron_connection.clone()).await;
omikron_connection.handle(&mut conn.receiver).await;
});
}