/* This Source Code Form is subject to the terms of the Mozilla Public * License, v. 2.0. If a copy of the MPL was not distributed with this * file, You can obtain one at http://mozilla.org/MPL/2.0/. */ //! A task that takes a URL and streams back the binary data. use about_loader; use cookie; use cookie_storage::CookieStorage; use data_loader; use devtools_traits::{DevtoolsControlMsg}; use file_loader; use hsts::{HSTSList, preload_hsts_domains}; use http_loader::{self, Connector, create_http_connector}; use hyper::client::pool::Pool; use hyper::header::{ContentType, Header, SetCookie}; use hyper::mime::{Mime, SubLevel, TopLevel}; use ipc_channel::ipc::{self, IpcReceiver, IpcSender}; use mime_classifier::{ApacheBugFlag, MIMEClassifier, NoSniffFlag}; use net_traits::LoadContext; use net_traits::ProgressMsg::Done; use net_traits::{AsyncResponseTarget, Metadata, ProgressMsg, ResourceTask, ResponseAction}; use net_traits::{ControlMsg, CookieSource, LoadConsumer, LoadData, LoadResponse, ResourceId}; use net_traits::{WebSocketCommunicate, WebSocketConnectData}; use std::borrow::ToOwned; use std::boxed::FnBox; use std::cell::Cell; use std::collections::HashMap; use std::sync::mpsc::{Receiver, Sender, channel}; use std::sync::{Arc, RwLock}; use url::Url; use util::opts; use util::task::spawn_named; use websocket_loader; pub enum ProgressSender { Channel(IpcSender), Listener(AsyncResponseTarget), } impl ProgressSender { //XXXjdm return actual error pub fn send(&self, msg: ProgressMsg) -> Result<(), ()> { match *self { ProgressSender::Channel(ref c) => c.send(msg).map_err(|_| ()), ProgressSender::Listener(ref b) => { let action = match msg { ProgressMsg::Payload(buf) => ResponseAction::DataAvailable(buf), ProgressMsg::Done(status) => ResponseAction::ResponseComplete(status), }; b.invoke_with_listener(action); Ok(()) } } } } pub fn send_error(url: Url, err: String, start_chan: LoadConsumer) { let mut metadata: Metadata = Metadata::default(url); metadata.status = None; if let Ok(p) = start_sending_opt(start_chan, metadata) { p.send(Done(Err(err))).unwrap(); } } /// For use by loaders in responding to a Load message that allows content sniffing. pub fn start_sending_sniffed(start_chan: LoadConsumer, metadata: Metadata, classifier: Arc, partial_body: &[u8], context: LoadContext) -> ProgressSender { start_sending_sniffed_opt(start_chan, metadata, classifier, partial_body, context).ok().unwrap() } /// For use by loaders in responding to a Load message that allows content sniffing. pub fn start_sending_sniffed_opt(start_chan: LoadConsumer, mut metadata: Metadata, classifier: Arc, partial_body: &[u8], context: LoadContext) -> Result { if opts::get().sniff_mime_types { // TODO: should be calculated in the resource loader, from pull requeset #4094 let mut no_sniff = NoSniffFlag::OFF; let mut check_for_apache_bug = ApacheBugFlag::OFF; if let Some(ref headers) = metadata.headers { if let Some(ref raw_content_type) = headers.get_raw("content-type") { if raw_content_type.len() > 0 { let last_raw_content_type = &raw_content_type[raw_content_type.len() - 1]; check_for_apache_bug = apache_bug_predicate(last_raw_content_type) } } if let Some(ref raw_content_type_options) = headers.get_raw("X-content-type-options") { if raw_content_type_options.iter().any(|ref opt| *opt == b"nosniff") { no_sniff = NoSniffFlag::ON } } } let supplied_type = metadata.content_type.as_ref().map(|&ContentType(Mime(ref toplevel, ref sublevel, _))| { (format!("{}", toplevel), format!("{}", sublevel)) }); let (toplevel, sublevel) = classifier.classify(context, no_sniff, check_for_apache_bug, &supplied_type, &partial_body); let mime_tp: TopLevel = toplevel.parse().unwrap(); let mime_sb: SubLevel = sublevel.parse().unwrap(); metadata.content_type = Some(ContentType(Mime(mime_tp, mime_sb, vec![]))); } start_sending_opt(start_chan, metadata) } fn apache_bug_predicate(last_raw_content_type: &[u8]) -> ApacheBugFlag { if last_raw_content_type == b"text/plain" || last_raw_content_type == b"text/plain; charset=ISO-8859-1" || last_raw_content_type == b"text/plain; charset=iso-8859-1" || last_raw_content_type == b"text/plain; charset=UTF-8" { ApacheBugFlag::ON } else { ApacheBugFlag::OFF } } /// For use by loaders in responding to a Load message. fn start_sending_opt(start_chan: LoadConsumer, metadata: Metadata) -> Result { match start_chan { LoadConsumer::Channel(start_chan) => { let (progress_chan, progress_port) = ipc::channel().unwrap(); let result = start_chan.send(LoadResponse { metadata: metadata, progress_port: progress_port, }); match result { Ok(_) => Ok(ProgressSender::Channel(progress_chan)), Err(_) => Err(()) } } LoadConsumer::Listener(target) => { target.invoke_with_listener(ResponseAction::HeadersAvailable(metadata)); Ok(ProgressSender::Listener(target)) } } } /// Create a ResourceTask pub fn new_resource_task(user_agent: String, devtools_chan: Option>) -> ResourceTask { let hsts_preload = match preload_hsts_domains() { Some(list) => list, None => HSTSList::new() }; let (setup_chan, setup_port) = ipc::channel().unwrap(); let setup_chan_clone = setup_chan.clone(); spawn_named("ResourceManager".to_owned(), move || { let resource_manager = ResourceManager::new( user_agent, hsts_preload, devtools_chan ); let mut channel_manager = ResourceChannelManager { from_client: setup_port, resource_manager: resource_manager }; channel_manager.start(setup_chan_clone); }); setup_chan } struct ResourceChannelManager { from_client: IpcReceiver, resource_manager: ResourceManager } impl ResourceChannelManager { fn start(&mut self, control_sender: ResourceTask) { loop { match self.from_client.recv().unwrap() { ControlMsg::Load(load_data, consumer, id_sender) => self.resource_manager.load(load_data, consumer, id_sender, control_sender.clone()), ControlMsg::WebsocketConnect(connect, connect_data) => self.resource_manager.websocket_connect(connect, connect_data), ControlMsg::SetCookiesForUrl(request, cookie_list, source) => self.resource_manager.set_cookies_for_url(request, cookie_list, source), ControlMsg::GetCookiesForUrl(url, consumer, source) => { let cookie_jar = &self.resource_manager.cookie_storage; let mut cookie_jar = cookie_jar.write().unwrap(); consumer.send(cookie_jar.cookies_for_url(&url, source)).unwrap(); } ControlMsg::Cancel(res_id) => { if let Some(cancel_sender) = self.resource_manager.cancel_load_map.get(&res_id) { let _ = cancel_sender.send(()); } self.resource_manager.cancel_load_map.remove(&res_id); } ControlMsg::Synchronize(sender) => { let _ = sender.send(()); } ControlMsg::Exit => break, } } } } /// The optional resources required by the `CancellationListener` pub struct CancellableResource { /// The receiver which receives a message on load cancellation cancel_receiver: Receiver<()>, /// The `CancellationListener` is unique to this `ResourceId` resource_id: ResourceId, /// If we haven't initiated any cancel requests, then the loaders ask /// the listener to remove the `ResourceId` in the `HashMap` of /// `ResourceManager` once they finish loading resource_task: ResourceTask, } impl CancellableResource { pub fn new(receiver: Receiver<()>, res_id: ResourceId, res_task: ResourceTask) -> CancellableResource { CancellableResource { cancel_receiver: receiver, resource_id: res_id, resource_task: res_task, } } } /// A listener which is basically a wrapped optional receiver which looks /// for the load cancellation message. Some of the loading processes always keep /// an eye out for this message and stop loading stuff once they receive it. pub struct CancellationListener { /// We'll be needing the resources only if we plan to cancel it cancel_resource: Option, /// This lets us know whether the request has already been cancelled cancel_status: Cell, } impl CancellationListener { pub fn new(resources: Option) -> CancellationListener { CancellationListener { cancel_resource: resources, cancel_status: Cell::new(false), } } pub fn is_cancelled(&self) -> bool { match self.cancel_resource { Some(ref resource) => { match resource.cancel_receiver.try_recv() { Ok(_) => { self.cancel_status.set(true); true }, Err(_) => self.cancel_status.get(), } }, None => false, // channel doesn't exist! } } } impl Drop for CancellationListener { fn drop(&mut self) { if let Some(ref resource) = self.cancel_resource { // Ensure that the resource manager stops tracking this request now that it's terminated. let _ = resource.resource_task.send(ControlMsg::Cancel(resource.resource_id)); } } } pub struct ResourceManager { user_agent: String, cookie_storage: Arc>, mime_classifier: Arc, devtools_chan: Option>, hsts_list: Arc>, connector: Arc>, cancel_load_map: HashMap>, next_resource_id: ResourceId, } impl ResourceManager { pub fn new(user_agent: String, hsts_list: HSTSList, devtools_channel: Option>) -> ResourceManager { ResourceManager { user_agent: user_agent, cookie_storage: Arc::new(RwLock::new(CookieStorage::new())), mime_classifier: Arc::new(MIMEClassifier::new()), devtools_chan: devtools_channel, hsts_list: Arc::new(RwLock::new(hsts_list)), connector: create_http_connector(), cancel_load_map: HashMap::new(), next_resource_id: ResourceId(0), } } fn set_cookies_for_url(&mut self, request: Url, cookie_list: String, source: CookieSource) { let header = Header::parse_header(&[cookie_list.into_bytes()]); if let Ok(SetCookie(cookies)) = header { for bare_cookie in cookies { if let Some(cookie) = cookie::Cookie::new_wrapped(bare_cookie, &request, source) { let cookie_jar = &self.cookie_storage; let mut cookie_jar = cookie_jar.write().unwrap(); cookie_jar.push(cookie, source); } } } } fn load(&mut self, load_data: LoadData, consumer: LoadConsumer, id_sender: Option>, resource_task: ResourceTask) { fn from_factory(factory: fn(LoadData, LoadConsumer, Arc, CancellationListener)) -> Box, CancellationListener) + Send> { box move |load_data, senders, classifier, cancel_listener| { factory(load_data, senders, classifier, cancel_listener) } } let cancel_resource = id_sender.map(|sender| { let current_res_id = self.next_resource_id; let _ = sender.send(current_res_id); let (cancel_sender, cancel_receiver) = channel(); self.cancel_load_map.insert(current_res_id, cancel_sender); self.next_resource_id.0 += 1; CancellableResource::new(cancel_receiver, current_res_id, resource_task) }); let cancel_listener = CancellationListener::new(cancel_resource); let loader = match &*load_data.url.scheme { "file" => from_factory(file_loader::factory), "http" | "https" | "view-source" => http_loader::factory(self.user_agent.clone(), self.hsts_list.clone(), self.cookie_storage.clone(), self.devtools_chan.clone(), self.connector.clone()), "data" => from_factory(data_loader::factory), "about" => from_factory(about_loader::factory), _ => { debug!("resource_task: no loader for scheme {}", load_data.url.scheme); send_error(load_data.url, "no loader for scheme".to_owned(), consumer); return } }; debug!("resource_task: loading url: {}", load_data.url.serialize()); loader.call_box((load_data, consumer, self.mime_classifier.clone(), cancel_listener)); } fn websocket_connect(&self, connect: WebSocketCommunicate, connect_data: WebSocketConnectData) { websocket_loader::init(connect, connect_data); } }